diff --git a/go.mod b/go.mod index 083f5808e2..84bdf9696d 100644 --- a/go.mod +++ b/go.mod @@ -51,6 +51,7 @@ require ( github.com/fatih/color v1.10.0 github.com/fernet/fernet-go v0.0.0-20180830025343-9eac43b88a5e github.com/flynn/go-shlex v0.0.0-20150515145356-3f9db97f8568 // indirect + github.com/frankban/quicktest v1.14.3 // indirect github.com/fsnotify/fsnotify v1.4.9 github.com/ghodss/yaml v1.0.0 github.com/gin-gonic/gin v1.7.7 @@ -97,6 +98,7 @@ require ( github.com/opentracing-contrib/go-observer v0.0.0-20170622124052-a52f23424492 // indirect github.com/opentracing/opentracing-go v1.0.2 // indirect github.com/openzipkin/zipkin-go-opentracing v0.3.4 // indirect + github.com/pierrec/lz4 v2.6.1+incompatible // indirect github.com/pierrec/lz4/v4 v4.1.12 github.com/pkg/errors v0.9.1 github.com/pquerna/otp v1.2.0 @@ -154,7 +156,7 @@ require ( yunion.io/x/ovsdb v0.0.0-20200526071744-27bf0940cbc7 yunion.io/x/pkg v1.0.0 yunion.io/x/s3cli v0.0.0-20190917004522-13ac36d8687e - yunion.io/x/sqlchemy v1.1.1 + yunion.io/x/sqlchemy v1.1.2-0.20220530125156-f4b471d64c78 yunion.io/x/structarg v0.0.0-20220312084958-9c6c79c7d1c6 ) diff --git a/go.sum b/go.sum index 48302cbd5d..e85175f910 100644 --- a/go.sum +++ b/go.sum @@ -199,6 +199,7 @@ github.com/coreos/go-semver v0.3.0 h1:wkHLiw0WNATZnSG7epLsujiMCgPAc9xhjJ4tgnAxmf github.com/coreos/go-semver v0.3.0/go.mod h1:nnelYz7RCh+5ahJtPPxZlU+153eP4D4r3EedlOD2RNk= github.com/coreos/go-systemd/v22 v22.3.2 h1:D9/bQk5vlXQFZ6Kwuu6zaiXJ9oTPe68++AzAJc1DzSI= github.com/coreos/go-systemd/v22 v22.3.2/go.mod h1:Y58oyj3AT4RCenI/lSvhwexgC+NSVTIJ3seZv2GcEnc= +github.com/creack/pty v1.1.9/go.mod h1:oKZEueFk5CKHvIhNR5MUki03XCEU+Q6VDXinZuGJ33E= github.com/creack/pty v1.1.11 h1:07n33Z8lZxZ2qwegKbObQohDhXDQxiMMz1NOUGYlesw= github.com/creack/pty v1.1.11/go.mod h1:oKZEueFk5CKHvIhNR5MUki03XCEU+Q6VDXinZuGJ33E= github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= @@ -241,6 +242,8 @@ github.com/fernet/fernet-go v0.0.0-20180830025343-9eac43b88a5e h1:P10tZmVD2XclAa github.com/fernet/fernet-go v0.0.0-20180830025343-9eac43b88a5e/go.mod h1:2H9hjfbpSMHwY503FclkV/lZTBh2YlOmLLSda12uL8c= github.com/flynn/go-shlex v0.0.0-20150515145356-3f9db97f8568 h1:BHsljHzVlRcyQhjrss6TZTdY2VfCqZPbv5k3iBFa2ZQ= github.com/flynn/go-shlex v0.0.0-20150515145356-3f9db97f8568/go.mod h1:xEzjJPgXI435gkrCt3MPfRiAkVrwSbHsst4LCFVfpJc= +github.com/frankban/quicktest v1.14.3 h1:FJKSZTDHjyhriyC81FLQ0LY93eSai0ZyR/ZIkd3ZUKE= +github.com/frankban/quicktest v1.14.3/go.mod h1:mgiwOwqx65TmIk1wJ6Q7wvnVMocbUorkibMOrVTHZps= github.com/fsnotify/fsnotify v1.4.7/go.mod h1:jwhsz4b93w/PPRr/qN1Yymfu8t87LnFCMoQvtojpjFo= github.com/fsnotify/fsnotify v1.4.9 h1:hsms1Qyu0jgnwNXIxa+/V/PDsU6CfLf6CNO8H7IWoS4= github.com/fsnotify/fsnotify v1.4.9/go.mod h1:znqG4EE+3YCdAaPaxE2ZRY/06pZUdp0tY4IgpuI1SZQ= @@ -349,8 +352,9 @@ github.com/google/go-cmp v0.4.1/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/ github.com/google/go-cmp v0.5.0/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE= github.com/google/go-cmp v0.5.1/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE= github.com/google/go-cmp v0.5.4/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE= -github.com/google/go-cmp v0.5.5 h1:Khx7svrCpmxxtHBq5j2mp/xVjsi8hQMfNLvJFAlrGgU= github.com/google/go-cmp v0.5.5/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE= +github.com/google/go-cmp v0.5.7 h1:81/ik6ipDQS2aGcBfIN5dHDB36BwrStyeAQquSYCV4o= +github.com/google/go-cmp v0.5.7/go.mod h1:n+brtR0CgQNWTVd5ZUFpTBC8YFBDLK/h/bpaJ8/DtOE= github.com/google/go-querystring v1.0.0 h1:Xkwi/a1rcvNg1PPYe5vI8GbeBY/jrVuDX5ASuANWTrk= github.com/google/go-querystring v1.0.0/go.mod h1:odCYkC5MyYFN7vkCjXpyrEuKhc/BUO6wN/zVPAxq5ck= github.com/google/gofuzz v1.0.0/go.mod h1:dBl0BpW6vV/+mYPU4Po3pmUjxk6FQPldtuIdl/M65Eg= @@ -453,11 +457,13 @@ github.com/konsorten/go-windows-terminal-sequences v1.0.3 h1:CE8S1cTafDpPvMhIxNJ github.com/konsorten/go-windows-terminal-sequences v1.0.3/go.mod h1:T0+1ngSBFLxvqU3pZ+m/2kptfBszLMUkC4ZK/EgS/cQ= github.com/kr/logfmt v0.0.0-20140226030751-b84e30acd515/go.mod h1:+0opPa2QZZtGFBFZlji/RkVcI2GknAs/DXo4wKdlNEc= github.com/kr/pretty v0.1.0/go.mod h1:dAy3ld7l9f0ibDNOQOHHMYYIIbhfbHSm3C4ZsoJORNo= -github.com/kr/pretty v0.2.0 h1:s5hAObm+yFO5uHYt5dYjxi2rXrsnmRpJx4OYvIWUaQs= github.com/kr/pretty v0.2.0/go.mod h1:ipq/a2n7PKx3OHsz4KJII5eveXtPO4qwEXGdVfWzfnI= +github.com/kr/pretty v0.3.0 h1:WgNl7dwNpEZ6jJ9k1snq4pZsg7DOEN8hP9Xw0Tsjwk0= +github.com/kr/pretty v0.3.0/go.mod h1:640gp4NfQd8pI5XOwp5fnNeVWj67G7CFk/SaSQn7NBk= github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ= -github.com/kr/text v0.1.0 h1:45sCR5RtlFHMR4UwH9sdQ5TC8v0qDQCHnXt+kaKSTVE= github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI= +github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY= +github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE= github.com/leodido/go-urn v1.2.0 h1:hpXL4XnriNwQ/ABnpepYM/1vCLWNDfUNts8dX3xTG6Y= github.com/leodido/go-urn v1.2.0/go.mod h1:+8+nEpDfqqsY+g338gtMEUOtuK+4dEMhiQEgxpxOKII= github.com/lestrrat-go/iter v0.0.0-20200422075355-fc1769541911 h1:FvnrqecqX4zT0wOIbYK1gNgTm0677INEWiFY8UEYggY= @@ -567,8 +573,9 @@ github.com/openzipkin/zipkin-go-opentracing v0.3.4/go.mod h1:js2AbwmHW0YD9DwIw2J github.com/peterbourgon/diskv v2.0.1+incompatible/go.mod h1:uqqh8zWWbv1HBMNONnaR/tNboyR3/BZd58JJSHlUSCU= github.com/philhofer/fwd v1.0.0 h1:UbZqGr5Y38ApvM/V/jEljVxwocdweyH+vmYvRPBnbqQ= github.com/philhofer/fwd v1.0.0/go.mod h1:gk3iGcWd9+svBvR0sR+KPcfE+RNWozjowpeBVG3ZVNU= -github.com/pierrec/lz4 v2.0.5+incompatible h1:2xWsjqPFWcplujydGg4WmhC/6fZqK42wMM8aXeqhl0I= github.com/pierrec/lz4 v2.0.5+incompatible/go.mod h1:pdkljMzZIN41W+lC3N2tnIh5sFi+IEE17M5jbnwPHcY= +github.com/pierrec/lz4 v2.6.1+incompatible h1:9UY3+iC23yxF0UfGaYrGplQ+79Rg+h/q9FV9ix19jjM= +github.com/pierrec/lz4 v2.6.1+incompatible/go.mod h1:pdkljMzZIN41W+lC3N2tnIh5sFi+IEE17M5jbnwPHcY= github.com/pierrec/lz4/v4 v4.1.12 h1:44l88ehTZAUGW4VlO1QC4zkilL99M6Y9MXNwEs0uzP8= github.com/pierrec/lz4/v4 v4.1.12/go.mod h1:gZWDp/Ze/IJXGXf23ltt2EXimqmTUXEy0GFuRQyBid4= github.com/pkg/diff v0.0.0-20200914180035-5b29258ca4f7/go.mod h1:zO8QMzTeZd5cpnIkz/Gn6iK0jDfGicM1nynOkkPIl28= @@ -611,6 +618,8 @@ github.com/rivo/uniseg v0.2.0 h1:S1pD9weZBuJdFmowNwbpi7BJ8TNftyUImj/0WQi72jY= github.com/rivo/uniseg v0.2.0/go.mod h1:J6wj4VEh+S6ZtnVlnTBMWIodfgj8LQOQFoIToxlJtxc= github.com/rogpeppe/fastuuid v1.2.0/go.mod h1:jVj6XXZzXRy/MSR5jhDC/2q6DgLz+nrA6LYCDYWNEvQ= github.com/rogpeppe/go-internal v1.3.0/go.mod h1:M8bDsm7K2OlrFYOpmOWEs/qY81heoFRclV5y23lUDJ4= +github.com/rogpeppe/go-internal v1.6.1 h1:/FiVV8dS/e+YqF2JvO3yXRFbBLTIuSDkuC7aBOAvL+k= +github.com/rogpeppe/go-internal v1.6.1/go.mod h1:xXDCJY+GAPziupqXw64V24skbSoqbTEfhy4qGm1nDQc= github.com/ryszard/goskiplist v0.0.0-20150312221310-2dfbae5fcf46 h1:GHRpF1pTW19a8tTFrMLUcfWwyC0pnifVo2ClaLq+hP8= github.com/ryszard/goskiplist v0.0.0-20150312221310-2dfbae5fcf46/go.mod h1:uAQ5PCi+MFsC7HjREoAz1BU+Mq60+05gifQSsHSDG/8= github.com/satori/go.uuid v1.2.0 h1:0uYX9dsZ2yD7q2RtLRtPSdGDWzjeM3TbMJP9utgA0ww= @@ -1130,7 +1139,7 @@ yunion.io/x/pkg v1.0.0/go.mod h1:nYOMflWG1wQC6301rGVglQva9bYbJxHThEJUxRAgMFc= yunion.io/x/s3cli v0.0.0-20190917004522-13ac36d8687e h1:v+EzIadodSwkdZ/7bremd7J8J50Cise/HCylsOJngmo= yunion.io/x/s3cli v0.0.0-20190917004522-13ac36d8687e/go.mod h1:0iFKpOs1y4lbCxeOmq3Xx/0AcQoewVPwj62eRluioEo= yunion.io/x/sqlchemy v1.0.1/go.mod h1:FTdwPdGhMgh4E+UFXc9klI1Ok34fMuybTT+jLhOaIjI= -yunion.io/x/sqlchemy v1.1.1 h1:FwWvGWesK6Tza/BzYwEZjv/Uy16lv3YyCaKZeIDp608= -yunion.io/x/sqlchemy v1.1.1/go.mod h1:LdxyRXxK6LklfUVW1my2fNapK1kcYRZh8nNWnILwHyM= +yunion.io/x/sqlchemy v1.1.2-0.20220530125156-f4b471d64c78 h1:vLfHc2MDc2judWBvYOmplHZzn+VXl+Rv6v1aKtwmVXE= +yunion.io/x/sqlchemy v1.1.2-0.20220530125156-f4b471d64c78/go.mod h1:LdxyRXxK6LklfUVW1my2fNapK1kcYRZh8nNWnILwHyM= yunion.io/x/structarg v0.0.0-20220312084958-9c6c79c7d1c6 h1:WuWXhY3DvhdRTzWCJ/kwt3Ss6KIq7+KqJwb+esvNGwU= yunion.io/x/structarg v0.0.0-20220312084958-9c6c79c7d1c6/go.mod h1:EP6NSv2C0zzqBDTKumv8hPWLb3XvgMZDHQRfyuOrQng= diff --git a/vendor/github.com/pierrec/lz4/.gitignore b/vendor/github.com/pierrec/lz4/.gitignore index e48bab32a5..5e98735047 100644 --- a/vendor/github.com/pierrec/lz4/.gitignore +++ b/vendor/github.com/pierrec/lz4/.gitignore @@ -30,4 +30,5 @@ Temporary Items # End of https://www.gitignore.io/api/macos -lz4c/lz4c +cmd/*/*exe +.idea \ No newline at end of file diff --git a/vendor/github.com/pierrec/lz4/.travis.yml b/vendor/github.com/pierrec/lz4/.travis.yml index b2c806d577..fd6c6db713 100644 --- a/vendor/github.com/pierrec/lz4/.travis.yml +++ b/vendor/github.com/pierrec/lz4/.travis.yml @@ -1,9 +1,13 @@ language: go +env: + - GO111MODULE=off + go: - - 1.8.x - 1.9.x - 1.10.x + - 1.11.x + - 1.12.x - master matrix: @@ -16,3 +20,5 @@ sudo: false script: - go test -v -cpu=2 - go test -v -cpu=2 -race + - go test -v -cpu=2 -tags noasm + - go test -v -cpu=2 -race -tags noasm diff --git a/vendor/github.com/pierrec/lz4/README.md b/vendor/github.com/pierrec/lz4/README.md index 50a10ee160..4ee388e81b 100644 --- a/vendor/github.com/pierrec/lz4/README.md +++ b/vendor/github.com/pierrec/lz4/README.md @@ -1,24 +1,90 @@ -[![godoc](https://godoc.org/github.com/pierrec/lz4?status.png)](https://godoc.org/github.com/pierrec/lz4) +# lz4 : LZ4 compression in pure Go -# lz4 -LZ4 compression and decompression in pure Go. +[![GoDoc](https://godoc.org/github.com/pierrec/lz4?status.svg)](https://godoc.org/github.com/pierrec/lz4) +[![Build Status](https://travis-ci.org/pierrec/lz4.svg?branch=master)](https://travis-ci.org/pierrec/lz4) +[![Go Report Card](https://goreportcard.com/badge/github.com/pierrec/lz4)](https://goreportcard.com/report/github.com/pierrec/lz4) +[![GitHub tag (latest SemVer)](https://img.shields.io/github/tag/pierrec/lz4.svg?style=social)](https://github.com/pierrec/lz4/tags) -## Usage +## Overview -```go -import "github.com/pierrec/lz4" +This package provides a streaming interface to [LZ4 data streams](http://fastcompression.blogspot.fr/2013/04/lz4-streaming-format-final.html) as well as low level compress and uncompress functions for LZ4 data blocks. +The implementation is based on the reference C [one](https://github.com/lz4/lz4). + +## Install + +Assuming you have the go toolchain installed: + +``` +go get github.com/pierrec/lz4 ``` -## Description -Package lz4 implements reading and writing lz4 compressed data (a frame), -as specified in http://fastcompression.blogspot.fr/2013/04/lz4-streaming-format-final.html. +There is a command line interface tool to compress and decompress LZ4 files. -This package is **compatible with the LZ4 frame format** although the block level compression -and decompression functions are exposed and are fully compatible with the lz4 block format -definition, they are low level and should not be used directly. +``` +go install github.com/pierrec/lz4/cmd/lz4c +``` -For a complete description of an lz4 compressed block, see: -http://fastcompression.blogspot.fr/2011/05/lz4-explained.html +Usage -See https://github.com/Cyan4973/lz4 for the reference C implementation. +``` +Usage of lz4c: + -version + print the program version +Subcommands: +Compress the given files or from stdin to stdout. +compress [arguments] [ ...] + -bc + enable block checksum + -l int + compression level (0=fastest) + -sc + disable stream checksum + -size string + block max size [64K,256K,1M,4M] (default "4M") + +Uncompress the given files or from stdin to stdout. +uncompress [arguments] [ ...] + +``` + + +## Example + +``` +// Compress and uncompress an input string. +s := "hello world" +r := strings.NewReader(s) + +// The pipe will uncompress the data from the writer. +pr, pw := io.Pipe() +zw := lz4.NewWriter(pw) +zr := lz4.NewReader(pr) + +go func() { + // Compress the input string. + _, _ = io.Copy(zw, r) + _ = zw.Close() // Make sure the writer is closed + _ = pw.Close() // Terminate the pipe +}() + +_, _ = io.Copy(os.Stdout, zr) + +// Output: +// hello world +``` + +## Contributing + +Contributions are very welcome for bug fixing, performance improvements...! + +- Open an issue with a proper description +- Send a pull request with appropriate test case(s) + +## Contributors + +Thanks to all [contributors](https://github.com/pierrec/lz4/graphs/contributors) so far! + +Special thanks to [@Zariel](https://github.com/Zariel) for his asm implementation of the decoder. + +Special thanks to [@klauspost](https://github.com/klauspost) for his work on optimizing the code. diff --git a/vendor/github.com/pierrec/lz4/block.go b/vendor/github.com/pierrec/lz4/block.go index ef24f17e57..664d9be580 100644 --- a/vendor/github.com/pierrec/lz4/block.go +++ b/vendor/github.com/pierrec/lz4/block.go @@ -2,21 +2,14 @@ package lz4 import ( "encoding/binary" - "errors" + "math/bits" + "sync" ) -var ( - // ErrInvalidSourceShortBuffer is returned by UncompressBlock or CompressBLock when a compressed - // block is corrupted or the destination buffer is not large enough for the uncompressed data. - ErrInvalidSourceShortBuffer = errors.New("lz4: invalid source or destination buffer too short") - // ErrInvalid is returned when reading an invalid LZ4 archive. - ErrInvalid = errors.New("lz4: bad magic number") -) - -// blockHash hashes 4 bytes into a value < winSize. -func blockHash(x uint32) uint32 { - const hasher uint32 = 2654435761 // Knuth multiplicative hash. - return x * hasher >> hashShift +// blockHash hashes the lower 6 bytes into a value < htSize. +func blockHash(x uint64) uint32 { + const prime6bytes = 227718039650203 + return uint32(((x << (64 - 48)) * prime6bytes) >> (64 - hashLog)) } // CompressBlockBound returns the maximum size of a given buffer of size n, when not compressible. @@ -30,137 +23,127 @@ func CompressBlockBound(n int) int { // The destination buffer must be sized appropriately. // // An error is returned if the source data is invalid or the destination buffer is too small. -func UncompressBlock(src, dst []byte) (si int, err error) { - defer func() { - // It is now faster to let the runtime panic and recover on out of bound slice access - // than checking indices as we go along. - if recover() != nil { - err = ErrInvalidSourceShortBuffer - } - }() - sn := len(src) - if sn == 0 { +func UncompressBlock(src, dst []byte) (int, error) { + if len(src) == 0 { return 0, nil } - var di int - - for { - // Literals and match lengths (token). - b := int(src[si]) - si++ - - // Literals. - if lLen := b >> 4; lLen > 0 { - if lLen == 0xF { - for src[si] == 0xFF { - lLen += 0xFF - si++ - } - lLen += int(src[si]) - si++ - } - i := si - si += lLen - di += copy(dst[di:], src[i:si]) - - if si >= sn { - return di, nil - } - } - - si++ - _ = src[si] // Bound check elimination. - offset := int(src[si-1]) | int(src[si])<<8 - si++ - - // Match. - mLen := b & 0xF - if mLen == 0xF { - for src[si] == 0xFF { - mLen += 0xFF - si++ - } - mLen += int(src[si]) - si++ - } - mLen += minMatch - - // Copy the match. - i := di - offset - if offset > 0 && mLen >= offset { - // Efficiently copy the match dst[di-offset:di] into the dst slice. - bytesToCopy := offset * (mLen / offset) - expanded := dst[i:] - for n := offset; n <= bytesToCopy+offset; n *= 2 { - copy(expanded[n:], expanded[:n]) - } - di += bytesToCopy - mLen -= bytesToCopy - } - di += copy(dst[di:], dst[i:i+mLen]) + if di := decodeBlock(dst, src); di >= 0 { + return di, nil } + return 0, ErrInvalidSourceShortBuffer } // CompressBlock compresses the source buffer into the destination one. // This is the fast version of LZ4 compression and also the default one. -// The size of hashTable must be at least 64Kb. // -// The size of the compressed data is returned. If it is 0 and no error, then the data is incompressible. +// The argument hashTable is scratch space for a hash table used by the +// compressor. If provided, it should have length at least 1<<16. If it is +// shorter (or nil), CompressBlock allocates its own hash table. +// +// The size of the compressed data is returned. +// +// If the destination buffer size is lower than CompressBlockBound and +// the compressed size is 0 and no error, then the data is incompressible. // // An error is returned if the destination buffer is too small. -func CompressBlock(src, dst []byte, hashTable []int) (di int, err error) { - defer func() { - if recover() != nil { - err = ErrInvalidSourceShortBuffer - } - }() +func CompressBlock(src, dst []byte, hashTable []int) (_ int, err error) { + defer recoverBlock(&err) - sn, dn := len(src)-mfLimit, len(dst) - if sn <= 0 || dn == 0 { - return 0, nil + // Return 0, nil only if the destination buffer size is < CompressBlockBound. + isNotCompressible := len(dst) < CompressBlockBound(len(src)) + + // adaptSkipLog sets how quickly the compressor begins skipping blocks when data is incompressible. + // This significantly speeds up incompressible data and usually has very small impact on compression. + // bytes to skip = 1 + (bytes since last match >> adaptSkipLog) + const adaptSkipLog = 7 + if len(hashTable) < htSize { + htIface := htPool.Get() + defer htPool.Put(htIface) + hashTable = (*(htIface).(*[htSize]int))[:] + } + // Prove to the compiler the table has at least htSize elements. + // The compiler can see that "uint32() >> hashShift" cannot be out of bounds. + hashTable = hashTable[:htSize] + + // si: Current position of the search. + // anchor: Position of the current literals. + var si, di, anchor int + sn := len(src) - mfLimit + if sn <= 0 { + goto lastLiterals } - var si int // Fast scan strategy: the hash table only stores the last 4 bytes sequences. - // const accInit = 1 << skipStrength - - anchor := si // Position of the current literals. - // acc := accInit // Variable step: improves performance on non-compressible data. - for si < sn { - // Hash the next 4 bytes (sequence)... - match := binary.LittleEndian.Uint32(src[si:]) + // Hash the next 6 bytes (sequence)... + match := binary.LittleEndian.Uint64(src[si:]) h := blockHash(match) + h2 := blockHash(match >> 8) + // We check a match at s, s+1 and s+2 and pick the first one we get. + // Checking 3 only requires us to load the source one. ref := hashTable[h] + ref2 := hashTable[h2] hashTable[h] = si - if ref >= sn { // Invalid reference (dirty hashtable). - si++ - continue - } + hashTable[h2] = si + 1 offset := si - ref + + // If offset <= 0 we got an old entry in the hash table. if offset <= 0 || offset >= winSize || // Out of window. - match != binary.LittleEndian.Uint32(src[ref:]) { // Hash collision on different matches. - // si += acc >> skipStrength - // acc++ - si++ - continue + uint32(match) != binary.LittleEndian.Uint32(src[ref:]) { // Hash collision on different matches. + // No match. Start calculating another hash. + // The processor can usually do this out-of-order. + h = blockHash(match >> 16) + ref = hashTable[h] + + // Check the second match at si+1 + si += 1 + offset = si - ref2 + + if offset <= 0 || offset >= winSize || + uint32(match>>8) != binary.LittleEndian.Uint32(src[ref2:]) { + // No match. Check the third match at si+2 + si += 1 + offset = si - ref + hashTable[h] = si + + if offset <= 0 || offset >= winSize || + uint32(match>>16) != binary.LittleEndian.Uint32(src[ref:]) { + // Skip one extra byte (at si+3) before we check 3 matches again. + si += 2 + (si-anchor)>>adaptSkipLog + continue + } + } } // Match found. - // acc = accInit lLen := si - anchor // Literal length. + // We already matched 4 bytes. + mLen := 4 - // Encode match length part 1. - si += minMatch - mLen := si // Match length has minMatch already. - // Find the longest match, first looking by batches of 8 bytes. - for si < sn && binary.LittleEndian.Uint64(src[si:]) == binary.LittleEndian.Uint64(src[si-offset:]) { - si += 8 + // Extend backwards if we can, reducing literals. + tOff := si - offset - 1 + for lLen > 0 && tOff >= 0 && src[si-1] == src[tOff] { + si-- + tOff-- + lLen-- + mLen++ } - // Then byte by byte. - for si < sn && src[si] == src[si-offset] { - si++ + + // Add the match length, so we continue search at the end. + // Use mLen to store the offset base. + si, mLen = si+mLen, si+minMatch + + // Find the longest match by looking by batches of 8 bytes. + for si+8 < sn { + x := binary.LittleEndian.Uint64(src[si:]) ^ binary.LittleEndian.Uint64(src[si-offset:]) + if x == 0 { + si += 8 + } else { + // Stop is first non-zero byte. + si += bits.TrailingZeros64(x) >> 3 + break + } } mLen = si - mLen @@ -186,7 +169,7 @@ func CompressBlock(src, dst []byte, hashTable []int) (di int, err error) { di++ // Literals. - copy(dst[di:], src[anchor:anchor+lLen]) + copy(dst[di:di+lLen], src[anchor:anchor+lLen]) di += lLen + 2 anchor = si @@ -203,9 +186,17 @@ func CompressBlock(src, dst []byte, hashTable []int) (di int, err error) { dst[di] = byte(mLen) di++ } + // Check if we can load next values. + if si >= sn { + break + } + // Hash match end-2 + h = blockHash(binary.LittleEndian.Uint64(src[si-2:])) + hashTable[h] = si - 2 } - if anchor == 0 { +lastLiterals: + if isNotCompressible && anchor == 0 { // Incompressible. return 0, nil } @@ -226,48 +217,68 @@ func CompressBlock(src, dst []byte, hashTable []int) (di int, err error) { di++ // Write the last literals. - if di >= anchor { + if isNotCompressible && di >= anchor { // Incompressible. return 0, nil } - di += copy(dst[di:], src[anchor:]) + di += copy(dst[di:di+len(src)-anchor], src[anchor:]) return di, nil } +// Pool of hash tables for CompressBlock. +var htPool = sync.Pool{ + New: func() interface{} { + return new([htSize]int) + }, +} + +// blockHash hashes 4 bytes into a value < winSize. +func blockHashHC(x uint32) uint32 { + const hasher uint32 = 2654435761 // Knuth multiplicative hash. + return x * hasher >> (32 - winSizeLog) +} + // CompressBlockHC compresses the source buffer src into the destination dst // with max search depth (use 0 or negative value for no max). // // CompressBlockHC compression ratio is better than CompressBlock but it is also slower. // -// The size of the compressed data is returned. If it is 0 and no error, then the data is not compressible. +// The size of the compressed data is returned. +// +// If the destination buffer size is lower than CompressBlockBound and +// the compressed size is 0 and no error, then the data is incompressible. // // An error is returned if the destination buffer is too small. -func CompressBlockHC(src, dst []byte, depth int) (di int, err error) { - defer func() { - if recover() != nil { - err = ErrInvalidSourceShortBuffer - } - }() +func CompressBlockHC(src, dst []byte, depth int) (_ int, err error) { + defer recoverBlock(&err) - sn, dn := len(src)-mfLimit, len(dst) - if sn <= 0 || dn == 0 { - return 0, nil - } - var si int + // Return 0, nil only if the destination buffer size is < CompressBlockBound. + isNotCompressible := len(dst) < CompressBlockBound(len(src)) + + // adaptSkipLog sets how quickly the compressor begins skipping blocks when data is incompressible. + // This significantly speeds up incompressible data and usually has very small impact on compression. + // bytes to skip = 1 + (bytes since last match >> adaptSkipLog) + const adaptSkipLog = 7 + + var si, di, anchor int // hashTable: stores the last position found for a given hash - // chaingTable: stores previous positions for a given hash + // chainTable: stores previous positions for a given hash var hashTable, chainTable [winSize]int if depth <= 0 { depth = winSize } - anchor := si + sn := len(src) - mfLimit + if sn <= 0 { + goto lastLiterals + } + for si < sn { // Hash the next 4 bytes (sequence). match := binary.LittleEndian.Uint32(src[si:]) - h := blockHash(match) + h := blockHashHC(match) // Follow the chain until out of window and give the longest match. mLen := 0 @@ -280,13 +291,17 @@ func CompressBlockHC(src, dst []byte, depth int) (di int, err error) { } ml := 0 // Compare the current position with a previous with the same hash. - for ml < sn-si && binary.LittleEndian.Uint64(src[next+ml:]) == binary.LittleEndian.Uint64(src[si+ml:]) { - ml += 8 + for ml < sn-si { + x := binary.LittleEndian.Uint64(src[next+ml:]) ^ binary.LittleEndian.Uint64(src[si+ml:]) + if x == 0 { + ml += 8 + } else { + // Stop is first non-zero byte. + ml += bits.TrailingZeros64(x) >> 3 + break + } } - for ml < sn-si && src[next+ml] == src[si+ml] { - ml++ - } - if ml+1 < minMatch || ml <= mLen { + if ml < minMatch || ml <= mLen { // Match too small (>adaptSkipLog continue } @@ -315,7 +330,7 @@ func CompressBlockHC(src, dst []byte, depth int) (di int, err error) { for si, ml := winStart, si+mLen; si < ml; { match >>= 8 match |= uint32(src[si+3]) << 24 - h := blockHash(match) + h := blockHashHC(match) chainTable[si&winMask] = hashTable[h] hashTable[h] = si si++ @@ -347,7 +362,7 @@ func CompressBlockHC(src, dst []byte, depth int) (di int, err error) { di++ // Literals. - copy(dst[di:], src[anchor:anchor+lLen]) + copy(dst[di:di+lLen], src[anchor:anchor+lLen]) di += lLen anchor = si @@ -366,12 +381,13 @@ func CompressBlockHC(src, dst []byte, depth int) (di int, err error) { } } - if anchor == 0 { + if isNotCompressible && anchor == 0 { // Incompressible. return 0, nil } // Last literals. +lastLiterals: lLen := len(src) - anchor if lLen < 0xF { dst[di] = byte(lLen << 4) @@ -388,10 +404,10 @@ func CompressBlockHC(src, dst []byte, depth int) (di int, err error) { di++ // Write the last literals. - if di >= anchor { + if isNotCompressible && di >= anchor { // Incompressible. return 0, nil } - di += copy(dst[di:], src[anchor:]) + di += copy(dst[di:di+len(src)-anchor], src[anchor:]) return di, nil } diff --git a/vendor/github.com/pierrec/lz4/decode_amd64.go b/vendor/github.com/pierrec/lz4/decode_amd64.go new file mode 100644 index 0000000000..43cc14fbe2 --- /dev/null +++ b/vendor/github.com/pierrec/lz4/decode_amd64.go @@ -0,0 +1,8 @@ +// +build !appengine +// +build gc +// +build !noasm + +package lz4 + +//go:noescape +func decodeBlock(dst, src []byte) int diff --git a/vendor/github.com/pierrec/lz4/decode_amd64.s b/vendor/github.com/pierrec/lz4/decode_amd64.s new file mode 100644 index 0000000000..20fef39759 --- /dev/null +++ b/vendor/github.com/pierrec/lz4/decode_amd64.s @@ -0,0 +1,375 @@ +// +build !appengine +// +build gc +// +build !noasm + +#include "textflag.h" + +// AX scratch +// BX scratch +// CX scratch +// DX token +// +// DI &dst +// SI &src +// R8 &dst + len(dst) +// R9 &src + len(src) +// R11 &dst +// R12 short output end +// R13 short input end +// func decodeBlock(dst, src []byte) int +// using 50 bytes of stack currently +TEXT ·decodeBlock(SB), NOSPLIT, $64-56 + MOVQ dst_base+0(FP), DI + MOVQ DI, R11 + MOVQ dst_len+8(FP), R8 + ADDQ DI, R8 + + MOVQ src_base+24(FP), SI + MOVQ src_len+32(FP), R9 + ADDQ SI, R9 + + // shortcut ends + // short output end + MOVQ R8, R12 + SUBQ $32, R12 + // short input end + MOVQ R9, R13 + SUBQ $16, R13 + +loop: + // for si < len(src) + CMPQ SI, R9 + JGE end + + // token := uint32(src[si]) + MOVBQZX (SI), DX + INCQ SI + + // lit_len = token >> 4 + // if lit_len > 0 + // CX = lit_len + MOVQ DX, CX + SHRQ $4, CX + + // if lit_len != 0xF + CMPQ CX, $0xF + JEQ lit_len_loop_pre + CMPQ DI, R12 + JGE lit_len_loop_pre + CMPQ SI, R13 + JGE lit_len_loop_pre + + // copy shortcut + + // A two-stage shortcut for the most common case: + // 1) If the literal length is 0..14, and there is enough space, + // enter the shortcut and copy 16 bytes on behalf of the literals + // (in the fast mode, only 8 bytes can be safely copied this way). + // 2) Further if the match length is 4..18, copy 18 bytes in a similar + // manner; but we ensure that there's enough space in the output for + // those 18 bytes earlier, upon entering the shortcut (in other words, + // there is a combined check for both stages). + + // copy literal + MOVOU (SI), X0 + MOVOU X0, (DI) + ADDQ CX, DI + ADDQ CX, SI + + MOVQ DX, CX + ANDQ $0xF, CX + + // The second stage: prepare for match copying, decode full info. + // If it doesn't work out, the info won't be wasted. + // offset := uint16(data[:2]) + MOVWQZX (SI), DX + ADDQ $2, SI + + MOVQ DI, AX + SUBQ DX, AX + CMPQ AX, DI + JGT err_short_buf + + // if we can't do the second stage then jump straight to read the + // match length, we already have the offset. + CMPQ CX, $0xF + JEQ match_len_loop_pre + CMPQ DX, $8 + JLT match_len_loop_pre + CMPQ AX, R11 + JLT err_short_buf + + // memcpy(op + 0, match + 0, 8); + MOVQ (AX), BX + MOVQ BX, (DI) + // memcpy(op + 8, match + 8, 8); + MOVQ 8(AX), BX + MOVQ BX, 8(DI) + // memcpy(op +16, match +16, 2); + MOVW 16(AX), BX + MOVW BX, 16(DI) + + ADDQ $4, DI // minmatch + ADDQ CX, DI + + // shortcut complete, load next token + JMP loop + +lit_len_loop_pre: + // if lit_len > 0 + CMPQ CX, $0 + JEQ offset + CMPQ CX, $0xF + JNE copy_literal + +lit_len_loop: + // for src[si] == 0xFF + CMPB (SI), $0xFF + JNE lit_len_finalise + + // bounds check src[si+1] + MOVQ SI, AX + ADDQ $1, AX + CMPQ AX, R9 + JGT err_short_buf + + // lit_len += 0xFF + ADDQ $0xFF, CX + INCQ SI + JMP lit_len_loop + +lit_len_finalise: + // lit_len += int(src[si]) + // si++ + MOVBQZX (SI), AX + ADDQ AX, CX + INCQ SI + +copy_literal: + // bounds check src and dst + MOVQ SI, AX + ADDQ CX, AX + CMPQ AX, R9 + JGT err_short_buf + + MOVQ DI, AX + ADDQ CX, AX + CMPQ AX, R8 + JGT err_short_buf + + // whats a good cut off to call memmove? + CMPQ CX, $16 + JGT memmove_lit + + // if len(dst[di:]) < 16 + MOVQ R8, AX + SUBQ DI, AX + CMPQ AX, $16 + JLT memmove_lit + + // if len(src[si:]) < 16 + MOVQ R9, AX + SUBQ SI, AX + CMPQ AX, $16 + JLT memmove_lit + + MOVOU (SI), X0 + MOVOU X0, (DI) + + JMP finish_lit_copy + +memmove_lit: + // memmove(to, from, len) + MOVQ DI, 0(SP) + MOVQ SI, 8(SP) + MOVQ CX, 16(SP) + // spill + MOVQ DI, 24(SP) + MOVQ SI, 32(SP) + MOVQ CX, 40(SP) // need len to inc SI, DI after + MOVB DX, 48(SP) + CALL runtime·memmove(SB) + + // restore registers + MOVQ 24(SP), DI + MOVQ 32(SP), SI + MOVQ 40(SP), CX + MOVB 48(SP), DX + + // recalc initial values + MOVQ dst_base+0(FP), R8 + MOVQ R8, R11 + ADDQ dst_len+8(FP), R8 + MOVQ src_base+24(FP), R9 + ADDQ src_len+32(FP), R9 + MOVQ R8, R12 + SUBQ $32, R12 + MOVQ R9, R13 + SUBQ $16, R13 + +finish_lit_copy: + ADDQ CX, SI + ADDQ CX, DI + + CMPQ SI, R9 + JGE end + +offset: + // CX := mLen + // free up DX to use for offset + MOVQ DX, CX + + MOVQ SI, AX + ADDQ $2, AX + CMPQ AX, R9 + JGT err_short_buf + + // offset + // DX := int(src[si]) | int(src[si+1])<<8 + MOVWQZX (SI), DX + ADDQ $2, SI + + // 0 offset is invalid + CMPQ DX, $0 + JEQ err_corrupt + + ANDB $0xF, CX + +match_len_loop_pre: + // if mlen != 0xF + CMPB CX, $0xF + JNE copy_match + +match_len_loop: + // for src[si] == 0xFF + // lit_len += 0xFF + CMPB (SI), $0xFF + JNE match_len_finalise + + // bounds check src[si+1] + MOVQ SI, AX + ADDQ $1, AX + CMPQ AX, R9 + JGT err_short_buf + + ADDQ $0xFF, CX + INCQ SI + JMP match_len_loop + +match_len_finalise: + // lit_len += int(src[si]) + // si++ + MOVBQZX (SI), AX + ADDQ AX, CX + INCQ SI + +copy_match: + // mLen += minMatch + ADDQ $4, CX + + // check we have match_len bytes left in dst + // di+match_len < len(dst) + MOVQ DI, AX + ADDQ CX, AX + CMPQ AX, R8 + JGT err_short_buf + + // DX = offset + // CX = match_len + // BX = &dst + (di - offset) + MOVQ DI, BX + SUBQ DX, BX + + // check BX is within dst + // if BX < &dst + CMPQ BX, R11 + JLT err_short_buf + + // if offset + match_len < di + MOVQ BX, AX + ADDQ CX, AX + CMPQ DI, AX + JGT copy_interior_match + + // AX := len(dst[:di]) + // MOVQ DI, AX + // SUBQ R11, AX + + // copy 16 bytes at a time + // if di-offset < 16 copy 16-(di-offset) bytes to di + // then do the remaining + +copy_match_loop: + // for match_len >= 0 + // dst[di] = dst[i] + // di++ + // i++ + MOVB (BX), AX + MOVB AX, (DI) + INCQ DI + INCQ BX + DECQ CX + + CMPQ CX, $0 + JGT copy_match_loop + + JMP loop + +copy_interior_match: + CMPQ CX, $16 + JGT memmove_match + + // if len(dst[di:]) < 16 + MOVQ R8, AX + SUBQ DI, AX + CMPQ AX, $16 + JLT memmove_match + + MOVOU (BX), X0 + MOVOU X0, (DI) + + ADDQ CX, DI + JMP loop + +memmove_match: + // memmove(to, from, len) + MOVQ DI, 0(SP) + MOVQ BX, 8(SP) + MOVQ CX, 16(SP) + // spill + MOVQ DI, 24(SP) + MOVQ SI, 32(SP) + MOVQ CX, 40(SP) // need len to inc SI, DI after + CALL runtime·memmove(SB) + + // restore registers + MOVQ 24(SP), DI + MOVQ 32(SP), SI + MOVQ 40(SP), CX + + // recalc initial values + MOVQ dst_base+0(FP), R8 + MOVQ R8, R11 // TODO: make these sensible numbers + ADDQ dst_len+8(FP), R8 + MOVQ src_base+24(FP), R9 + ADDQ src_len+32(FP), R9 + MOVQ R8, R12 + SUBQ $32, R12 + MOVQ R9, R13 + SUBQ $16, R13 + + ADDQ CX, DI + JMP loop + +err_corrupt: + MOVQ $-1, ret+48(FP) + RET + +err_short_buf: + MOVQ $-2, ret+48(FP) + RET + +end: + SUBQ R11, DI + MOVQ DI, ret+48(FP) + RET diff --git a/vendor/github.com/pierrec/lz4/decode_other.go b/vendor/github.com/pierrec/lz4/decode_other.go new file mode 100644 index 0000000000..919888edf7 --- /dev/null +++ b/vendor/github.com/pierrec/lz4/decode_other.go @@ -0,0 +1,98 @@ +// +build !amd64 appengine !gc noasm + +package lz4 + +func decodeBlock(dst, src []byte) (ret int) { + const hasError = -2 + defer func() { + if recover() != nil { + ret = hasError + } + }() + + var si, di int + for { + // Literals and match lengths (token). + b := int(src[si]) + si++ + + // Literals. + if lLen := b >> 4; lLen > 0 { + switch { + case lLen < 0xF && si+16 < len(src): + // Shortcut 1 + // if we have enough room in src and dst, and the literals length + // is small enough (0..14) then copy all 16 bytes, even if not all + // are part of the literals. + copy(dst[di:], src[si:si+16]) + si += lLen + di += lLen + if mLen := b & 0xF; mLen < 0xF { + // Shortcut 2 + // if the match length (4..18) fits within the literals, then copy + // all 18 bytes, even if not all are part of the literals. + mLen += 4 + if offset := int(src[si]) | int(src[si+1])<<8; mLen <= offset { + i := di - offset + end := i + 18 + if end > len(dst) { + // The remaining buffer may not hold 18 bytes. + // See https://github.com/pierrec/lz4/issues/51. + end = len(dst) + } + copy(dst[di:], dst[i:end]) + si += 2 + di += mLen + continue + } + } + case lLen == 0xF: + for src[si] == 0xFF { + lLen += 0xFF + si++ + } + lLen += int(src[si]) + si++ + fallthrough + default: + copy(dst[di:di+lLen], src[si:si+lLen]) + si += lLen + di += lLen + } + } + if si >= len(src) { + return di + } + + offset := int(src[si]) | int(src[si+1])<<8 + if offset == 0 { + return hasError + } + si += 2 + + // Match. + mLen := b & 0xF + if mLen == 0xF { + for src[si] == 0xFF { + mLen += 0xFF + si++ + } + mLen += int(src[si]) + si++ + } + mLen += minMatch + + // Copy the match. + expanded := dst[di-offset:] + if mLen > offset { + // Efficiently copy the match dst[di-offset:di] into the dst slice. + bytesToCopy := offset * (mLen / offset) + for n := offset; n <= bytesToCopy+offset; n *= 2 { + copy(expanded[n:], expanded[:n]) + } + di += bytesToCopy + mLen -= bytesToCopy + } + di += copy(dst[di:di+mLen], expanded[:mLen]) + } +} diff --git a/vendor/github.com/pierrec/lz4/errors.go b/vendor/github.com/pierrec/lz4/errors.go new file mode 100644 index 0000000000..1c45d1813c --- /dev/null +++ b/vendor/github.com/pierrec/lz4/errors.go @@ -0,0 +1,30 @@ +package lz4 + +import ( + "errors" + "fmt" + "os" + rdebug "runtime/debug" +) + +var ( + // ErrInvalidSourceShortBuffer is returned by UncompressBlock or CompressBLock when a compressed + // block is corrupted or the destination buffer is not large enough for the uncompressed data. + ErrInvalidSourceShortBuffer = errors.New("lz4: invalid source or destination buffer too short") + // ErrInvalid is returned when reading an invalid LZ4 archive. + ErrInvalid = errors.New("lz4: bad magic number") + // ErrBlockDependency is returned when attempting to decompress an archive created with block dependency. + ErrBlockDependency = errors.New("lz4: block dependency not supported") + // ErrUnsupportedSeek is returned when attempting to Seek any way but forward from the current position. + ErrUnsupportedSeek = errors.New("lz4: can only seek forward from io.SeekCurrent") +) + +func recoverBlock(e *error) { + if r := recover(); r != nil && *e == nil { + if debugFlag { + fmt.Fprintln(os.Stderr, r) + rdebug.PrintStack() + } + *e = ErrInvalidSourceShortBuffer + } +} diff --git a/vendor/github.com/pierrec/lz4/internal/xxh32/xxh32zero.go b/vendor/github.com/pierrec/lz4/internal/xxh32/xxh32zero.go index 850a6fdf61..7a76a6bce2 100644 --- a/vendor/github.com/pierrec/lz4/internal/xxh32/xxh32zero.go +++ b/vendor/github.com/pierrec/lz4/internal/xxh32/xxh32zero.go @@ -7,14 +7,15 @@ import ( ) const ( - prime32_1 uint32 = 2654435761 - prime32_2 uint32 = 2246822519 - prime32_3 uint32 = 3266489917 - prime32_4 uint32 = 668265263 - prime32_5 uint32 = 374761393 + prime1 uint32 = 2654435761 + prime2 uint32 = 2246822519 + prime3 uint32 = 3266489917 + prime4 uint32 = 668265263 + prime5 uint32 = 374761393 - prime32_1plus2 uint32 = 606290984 - prime32_minus1 uint32 = 1640531535 + primeMask = 0xFFFFFFFF + prime1plus2 = uint32((uint64(prime1) + uint64(prime2)) & primeMask) // 606290984 + prime1minus = uint32((-int64(prime1)) & primeMask) // 1640531535 ) // XXHZero represents an xxhash32 object with seed 0. @@ -37,10 +38,10 @@ func (xxh XXHZero) Sum(b []byte) []byte { // Reset resets the Hash to its initial state. func (xxh *XXHZero) Reset() { - xxh.v1 = prime32_1plus2 - xxh.v2 = prime32_2 + xxh.v1 = prime1plus2 + xxh.v2 = prime2 xxh.v3 = 0 - xxh.v4 = prime32_minus1 + xxh.v4 = prime1minus xxh.totalLen = 0 xxh.bufused = 0 } @@ -83,20 +84,20 @@ func (xxh *XXHZero) Write(input []byte) (int, error) { // fast rotl(13) buf := xxh.buf[:16] // BCE hint. - v1 = rol13(v1+binary.LittleEndian.Uint32(buf[:])*prime32_2) * prime32_1 - v2 = rol13(v2+binary.LittleEndian.Uint32(buf[4:])*prime32_2) * prime32_1 - v3 = rol13(v3+binary.LittleEndian.Uint32(buf[8:])*prime32_2) * prime32_1 - v4 = rol13(v4+binary.LittleEndian.Uint32(buf[12:])*prime32_2) * prime32_1 + v1 = rol13(v1+binary.LittleEndian.Uint32(buf[:])*prime2) * prime1 + v2 = rol13(v2+binary.LittleEndian.Uint32(buf[4:])*prime2) * prime1 + v3 = rol13(v3+binary.LittleEndian.Uint32(buf[8:])*prime2) * prime1 + v4 = rol13(v4+binary.LittleEndian.Uint32(buf[12:])*prime2) * prime1 p = r xxh.bufused = 0 } for n := n - 16; p <= n; p += 16 { sub := input[p:][:16] //BCE hint for compiler - v1 = rol13(v1+binary.LittleEndian.Uint32(sub[:])*prime32_2) * prime32_1 - v2 = rol13(v2+binary.LittleEndian.Uint32(sub[4:])*prime32_2) * prime32_1 - v3 = rol13(v3+binary.LittleEndian.Uint32(sub[8:])*prime32_2) * prime32_1 - v4 = rol13(v4+binary.LittleEndian.Uint32(sub[12:])*prime32_2) * prime32_1 + v1 = rol13(v1+binary.LittleEndian.Uint32(sub[:])*prime2) * prime1 + v2 = rol13(v2+binary.LittleEndian.Uint32(sub[4:])*prime2) * prime1 + v3 = rol13(v3+binary.LittleEndian.Uint32(sub[8:])*prime2) * prime1 + v4 = rol13(v4+binary.LittleEndian.Uint32(sub[12:])*prime2) * prime1 } xxh.v1, xxh.v2, xxh.v3, xxh.v4 = v1, v2, v3, v4 @@ -112,25 +113,25 @@ func (xxh *XXHZero) Sum32() uint32 { if h32 >= 16 { h32 += rol1(xxh.v1) + rol7(xxh.v2) + rol12(xxh.v3) + rol18(xxh.v4) } else { - h32 += prime32_5 + h32 += prime5 } p := 0 n := xxh.bufused buf := xxh.buf for n := n - 4; p <= n; p += 4 { - h32 += binary.LittleEndian.Uint32(buf[p:p+4]) * prime32_3 - h32 = rol17(h32) * prime32_4 + h32 += binary.LittleEndian.Uint32(buf[p:p+4]) * prime3 + h32 = rol17(h32) * prime4 } for ; p < n; p++ { - h32 += uint32(buf[p]) * prime32_5 - h32 = rol11(h32) * prime32_1 + h32 += uint32(buf[p]) * prime5 + h32 = rol11(h32) * prime1 } h32 ^= h32 >> 15 - h32 *= prime32_2 + h32 *= prime2 h32 ^= h32 >> 13 - h32 *= prime32_3 + h32 *= prime3 h32 ^= h32 >> 16 return h32 @@ -142,19 +143,19 @@ func ChecksumZero(input []byte) uint32 { h32 := uint32(n) if n < 16 { - h32 += prime32_5 + h32 += prime5 } else { - v1 := prime32_1plus2 - v2 := prime32_2 + v1 := prime1plus2 + v2 := prime2 v3 := uint32(0) - v4 := prime32_minus1 + v4 := prime1minus p := 0 for n := n - 16; p <= n; p += 16 { sub := input[p:][:16] //BCE hint for compiler - v1 = rol13(v1+binary.LittleEndian.Uint32(sub[:])*prime32_2) * prime32_1 - v2 = rol13(v2+binary.LittleEndian.Uint32(sub[4:])*prime32_2) * prime32_1 - v3 = rol13(v3+binary.LittleEndian.Uint32(sub[8:])*prime32_2) * prime32_1 - v4 = rol13(v4+binary.LittleEndian.Uint32(sub[12:])*prime32_2) * prime32_1 + v1 = rol13(v1+binary.LittleEndian.Uint32(sub[:])*prime2) * prime1 + v2 = rol13(v2+binary.LittleEndian.Uint32(sub[4:])*prime2) * prime1 + v3 = rol13(v3+binary.LittleEndian.Uint32(sub[8:])*prime2) * prime1 + v4 = rol13(v4+binary.LittleEndian.Uint32(sub[12:])*prime2) * prime1 } input = input[p:] n -= p @@ -163,19 +164,19 @@ func ChecksumZero(input []byte) uint32 { p := 0 for n := n - 4; p <= n; p += 4 { - h32 += binary.LittleEndian.Uint32(input[p:p+4]) * prime32_3 - h32 = rol17(h32) * prime32_4 + h32 += binary.LittleEndian.Uint32(input[p:p+4]) * prime3 + h32 = rol17(h32) * prime4 } for p < n { - h32 += uint32(input[p]) * prime32_5 - h32 = rol11(h32) * prime32_1 + h32 += uint32(input[p]) * prime5 + h32 = rol11(h32) * prime1 p++ } h32 ^= h32 >> 15 - h32 *= prime32_2 + h32 *= prime2 h32 ^= h32 >> 13 - h32 *= prime32_3 + h32 *= prime3 h32 ^= h32 >> 16 return h32 @@ -183,12 +184,12 @@ func ChecksumZero(input []byte) uint32 { // Uint32Zero hashes x with seed 0. func Uint32Zero(x uint32) uint32 { - h := prime32_5 + 4 + x*prime32_3 - h = rol17(h) * prime32_4 + h := prime5 + 4 + x*prime3 + h = rol17(h) * prime4 h ^= h >> 15 - h *= prime32_2 + h *= prime2 h ^= h >> 13 - h *= prime32_3 + h *= prime3 h ^= h >> 16 return h } diff --git a/vendor/github.com/pierrec/lz4/lz4.go b/vendor/github.com/pierrec/lz4/lz4.go index 35802756c4..a3284bdf70 100644 --- a/vendor/github.com/pierrec/lz4/lz4.go +++ b/vendor/github.com/pierrec/lz4/lz4.go @@ -10,14 +10,20 @@ // package lz4 +import ( + "math/bits" + "sync" +) + const ( // Extension is the LZ4 frame file name extension Extension = ".lz4" // Version is the LZ4 frame format version Version = 1 - frameMagic uint32 = 0x184D2204 - frameSkipMagic uint32 = 0x184D2A50 + frameMagic uint32 = 0x184D2204 + frameSkipMagic uint32 = 0x184D2A50 + frameMagicLegacy uint32 = 0x184C2102 // The following constants are used to setup the compression algorithm. minMatch = 4 // the minimum size of the match sequence size (4 bytes) @@ -30,26 +36,63 @@ const ( // hashLog determines the size of the hash table used to quickly find a previous match position. // Its value influences the compression speed and memory usage, the lower the faster, // but at the expense of the compression ratio. - // 16 seems to be the best compromise. - hashLog = 16 - hashTableSize = 1 << hashLog - hashShift = uint((minMatch * 8) - hashLog) + // 16 seems to be the best compromise for fast compression. + hashLog = 16 + htSize = 1 << hashLog - mfLimit = 8 + minMatch // The last match cannot start within the last 12 bytes. - skipStrength = 6 // variable step for fast scan + mfLimit = 10 + minMatch // The last match cannot start within the last 14 bytes. ) // map the block max size id with its value in bytes: 64Kb, 256Kb, 1Mb and 4Mb. -var ( - bsMapID = map[byte]int{4: 64 << 10, 5: 256 << 10, 6: 1 << 20, 7: 4 << 20} - bsMapValue = make(map[int]byte, len(bsMapID)) +const ( + blockSize64K = 1 << (16 + 2*iota) + blockSize256K + blockSize1M + blockSize4M ) -// Reversed. -func init() { - for i, v := range bsMapID { - bsMapValue[v] = i +var ( + // Keep a pool of buffers for each valid block sizes. + bsMapValue = [...]*sync.Pool{ + newBufferPool(2 * blockSize64K), + newBufferPool(2 * blockSize256K), + newBufferPool(2 * blockSize1M), + newBufferPool(2 * blockSize4M), } +) + +// newBufferPool returns a pool for buffers of the given size. +func newBufferPool(size int) *sync.Pool { + return &sync.Pool{ + New: func() interface{} { + return make([]byte, size) + }, + } +} + +// getBuffer returns a buffer to its pool. +func getBuffer(size int) []byte { + idx := blockSizeValueToIndex(size) - 4 + return bsMapValue[idx].Get().([]byte) +} + +// putBuffer returns a buffer to its pool. +func putBuffer(size int, buf []byte) { + if cap(buf) > 0 { + idx := blockSizeValueToIndex(size) - 4 + bsMapValue[idx].Put(buf[:cap(buf)]) + } +} +func blockSizeIndexToValue(i byte) int { + return 1 << (16 + 2*uint(i)) +} +func isValidBlockSize(size int) bool { + const blockSizeMask = blockSize64K | blockSize256K | blockSize1M | blockSize4M + + return size&blockSizeMask > 0 && bits.OnesCount(uint(size)) == 1 +} +func blockSizeValueToIndex(size int) byte { + return 4 + byte(bits.TrailingZeros(uint(size)>>16)/2) } // Header describes the various flags that can be set on a Writer or obtained from a Reader. @@ -57,7 +100,7 @@ func init() { // (http://fastcompression.blogspot.com/2013/04/lz4-streaming-format-final.html). // // NB. in a Reader, in case of concatenated frames, the Header values may change between Read() calls. -// It is the caller responsibility to check them if necessary. +// It is the caller's responsibility to check them if necessary. type Header struct { BlockChecksum bool // Compressed blocks checksum flag. NoChecksum bool // Frame checksum flag. @@ -66,3 +109,8 @@ type Header struct { CompressionLevel int // Compression level (higher is better, use 0 for fastest compression). done bool // Header processed flag (Read or Write and checked). } + +// Reset reset internal status +func (h *Header) Reset() { + h.done = false +} diff --git a/vendor/github.com/pierrec/lz4/reader.go b/vendor/github.com/pierrec/lz4/reader.go index f08db47df7..87dd72bd0d 100644 --- a/vendor/github.com/pierrec/lz4/reader.go +++ b/vendor/github.com/pierrec/lz4/reader.go @@ -14,6 +14,9 @@ import ( // The Header may change between Read() calls in case of concatenated frames. type Reader struct { Header + // Handler called when a block has been successfully read. + // It provides the number of bytes read. + OnBlockDone func(size int) buf [8]byte // Scrap buffer. pos int64 // Current position in src. @@ -22,6 +25,8 @@ type Reader struct { data []byte // Uncompressed data. idx int // Index of unread bytes into data. checksum xxh32.XXHZero // Frame hash. + skip int64 // Bytes to skip before next read. + dpos int64 // Position in dest } // NewReader returns a new LZ4 frame decoder. @@ -76,17 +81,17 @@ func (z *Reader) readHeader(first bool) error { return fmt.Errorf("lz4: invalid version: got %d; expected %d", v, Version) } if b>>5&1 == 0 { - return fmt.Errorf("lz4: block dependency not supported") + return ErrBlockDependency } z.BlockChecksum = b>>4&1 > 0 frameSize := b>>3&1 > 0 z.NoChecksum = b>>2&1 == 0 bmsID := buf[1] >> 4 & 0x7 - bSize, ok := bsMapID[bmsID] - if !ok { + if bmsID < 4 || bmsID > 7 { return fmt.Errorf("lz4: invalid block max size ID: %d", bmsID) } + bSize := blockSizeIndexToValue(bmsID - 4) z.BlockMaxSize = bSize // Allocate the compressed/uncompressed buffers. @@ -101,7 +106,7 @@ func (z *Reader) readHeader(first bool) error { z.data = z.zdata[:cap(z.zdata)][bSize:] z.idx = len(z.data) - z.checksum.Write(buf[0:2]) + _, _ = z.checksum.Write(buf[0:2]) if frameSize { buf := buf[:8] @@ -110,7 +115,7 @@ func (z *Reader) readHeader(first bool) error { } z.Size = binary.LittleEndian.Uint64(buf) z.pos += 8 - z.checksum.Write(buf) + _, _ = z.checksum.Write(buf) } // Header checksum. @@ -158,6 +163,9 @@ func (z *Reader) Read(buf []byte) (int, error) { if debugFlag { debug("reading block from writer") } + // Reset uncompressed buffer + z.data = z.zdata[:cap(z.zdata)][len(z.zdata):] + // Block length: 0 = end of frame, highest bit set: uncompressed. bLen, err := z.readUint32() if err != nil { @@ -208,6 +216,9 @@ func (z *Reader) Read(buf []byte) (int, error) { return 0, err } z.pos += int64(bLen) + if z.OnBlockDone != nil { + z.OnBlockDone(int(bLen)) + } if z.BlockChecksum { checksum, err := z.readUint32() @@ -252,10 +263,13 @@ func (z *Reader) Read(buf []byte) (int, error) { return 0, err } z.data = z.data[:n] + if z.OnBlockDone != nil { + z.OnBlockDone(n) + } } if !z.NoChecksum { - z.checksum.Write(z.data) + _, _ = z.checksum.Write(z.data) if debugFlag { debug("current frame checksum %x", z.checksum.Sum32()) } @@ -263,8 +277,20 @@ func (z *Reader) Read(buf []byte) (int, error) { z.idx = 0 } + if z.skip > int64(len(z.data[z.idx:])) { + z.skip -= int64(len(z.data[z.idx:])) + z.dpos += int64(len(z.data[z.idx:])) + z.idx = len(z.data) + return 0, nil + } + + z.idx += int(z.skip) + z.dpos += z.skip + z.skip = 0 + n := copy(buf, z.data[z.idx:]) z.idx += n + z.dpos += int64(n) if debugFlag { debug("copied %d bytes to input", n) } @@ -272,6 +298,20 @@ func (z *Reader) Read(buf []byte) (int, error) { return n, nil } +// Seek implements io.Seeker, but supports seeking forward from the current +// position only. Any other seek will return an error. Allows skipping output +// bytes which aren't needed, which in some scenarios is faster than reading +// and discarding them. +// Note this may cause future calls to Read() to read 0 bytes if all of the +// data they would have returned is skipped. +func (z *Reader) Seek(offset int64, whence int) (int64, error) { + if offset < 0 || whence != io.SeekCurrent { + return z.dpos + z.skip, ErrUnsupportedSeek + } + z.skip += offset + return z.dpos + z.skip, nil +} + // Reset discards the Reader's state and makes it equivalent to the // result of its original state from NewReader, but reading from r instead. // This permits reusing a Reader rather than allocating a new one. diff --git a/vendor/github.com/pierrec/lz4/reader_legacy.go b/vendor/github.com/pierrec/lz4/reader_legacy.go new file mode 100644 index 0000000000..1670a77d02 --- /dev/null +++ b/vendor/github.com/pierrec/lz4/reader_legacy.go @@ -0,0 +1,207 @@ +package lz4 + +import ( + "encoding/binary" + "fmt" + "io" +) + +// ReaderLegacy implements the LZ4Demo frame decoder. +// The Header is set after the first call to Read(). +type ReaderLegacy struct { + Header + // Handler called when a block has been successfully read. + // It provides the number of bytes read. + OnBlockDone func(size int) + + lastBlock bool + buf [8]byte // Scrap buffer. + pos int64 // Current position in src. + src io.Reader // Source. + zdata []byte // Compressed data. + data []byte // Uncompressed data. + idx int // Index of unread bytes into data. + skip int64 // Bytes to skip before next read. + dpos int64 // Position in dest +} + +// NewReaderLegacy returns a new LZ4Demo frame decoder. +// No access to the underlying io.Reader is performed. +func NewReaderLegacy(src io.Reader) *ReaderLegacy { + r := &ReaderLegacy{src: src} + return r +} + +// readHeader checks the frame magic number and parses the frame descriptoz. +// Skippable frames are supported even as a first frame although the LZ4 +// specifications recommends skippable frames not to be used as first frames. +func (z *ReaderLegacy) readLegacyHeader() error { + z.lastBlock = false + magic, err := z.readUint32() + if err != nil { + z.pos += 4 + if err == io.ErrUnexpectedEOF { + return io.EOF + } + return err + } + if magic != frameMagicLegacy { + return ErrInvalid + } + z.pos += 4 + + // Legacy has fixed 8MB blocksizes + // https://github.com/lz4/lz4/blob/dev/doc/lz4_Frame_format.md#legacy-frame + bSize := blockSize4M * 2 + + // Allocate the compressed/uncompressed buffers. + // The compressed buffer cannot exceed the uncompressed one. + if n := 2 * bSize; cap(z.zdata) < n { + z.zdata = make([]byte, n, n) + } + if debugFlag { + debug("header block max size size=%d", bSize) + } + z.zdata = z.zdata[:bSize] + z.data = z.zdata[:cap(z.zdata)][bSize:] + z.idx = len(z.data) + + z.Header.done = true + if debugFlag { + debug("header read: %v", z.Header) + } + + return nil +} + +// Read decompresses data from the underlying source into the supplied buffer. +// +// Since there can be multiple streams concatenated, Header values may +// change between calls to Read(). If that is the case, no data is actually read from +// the underlying io.Reader, to allow for potential input buffer resizing. +func (z *ReaderLegacy) Read(buf []byte) (int, error) { + if debugFlag { + debug("Read buf len=%d", len(buf)) + } + if !z.Header.done { + if err := z.readLegacyHeader(); err != nil { + return 0, err + } + if debugFlag { + debug("header read OK compressed buffer %d / %d uncompressed buffer %d : %d index=%d", + len(z.zdata), cap(z.zdata), len(z.data), cap(z.data), z.idx) + } + } + + if len(buf) == 0 { + return 0, nil + } + + if z.idx == len(z.data) { + // No data ready for reading, process the next block. + if debugFlag { + debug(" reading block from writer %d %d", z.idx, blockSize4M*2) + } + + // Reset uncompressed buffer + z.data = z.zdata[:cap(z.zdata)][len(z.zdata):] + + bLen, err := z.readUint32() + if err != nil { + return 0, err + } + if debugFlag { + debug(" bLen %d (0x%x) offset = %d (0x%x)", bLen, bLen, z.pos, z.pos) + } + z.pos += 4 + + // Legacy blocks are always compressed, even when detrimental + if debugFlag { + debug(" compressed block size %d", bLen) + } + + if int(bLen) > cap(z.data) { + return 0, fmt.Errorf("lz4: invalid block size: %d", bLen) + } + zdata := z.zdata[:bLen] + if _, err := io.ReadFull(z.src, zdata); err != nil { + return 0, err + } + z.pos += int64(bLen) + + n, err := UncompressBlock(zdata, z.data) + if err != nil { + return 0, err + } + + z.data = z.data[:n] + if z.OnBlockDone != nil { + z.OnBlockDone(n) + } + + z.idx = 0 + + // Legacy blocks are fixed to 8MB, if we read a decompressed block smaller than this + // it means we've reached the end... + if n < blockSize4M*2 { + z.lastBlock = true + } + } + + if z.skip > int64(len(z.data[z.idx:])) { + z.skip -= int64(len(z.data[z.idx:])) + z.dpos += int64(len(z.data[z.idx:])) + z.idx = len(z.data) + return 0, nil + } + + z.idx += int(z.skip) + z.dpos += z.skip + z.skip = 0 + + n := copy(buf, z.data[z.idx:]) + z.idx += n + z.dpos += int64(n) + if debugFlag { + debug("%v] copied %d bytes to input (%d:%d)", z.lastBlock, n, z.idx, len(z.data)) + } + if z.lastBlock && len(z.data) == z.idx { + return n, io.EOF + } + return n, nil +} + +// Seek implements io.Seeker, but supports seeking forward from the current +// position only. Any other seek will return an error. Allows skipping output +// bytes which aren't needed, which in some scenarios is faster than reading +// and discarding them. +// Note this may cause future calls to Read() to read 0 bytes if all of the +// data they would have returned is skipped. +func (z *ReaderLegacy) Seek(offset int64, whence int) (int64, error) { + if offset < 0 || whence != io.SeekCurrent { + return z.dpos + z.skip, ErrUnsupportedSeek + } + z.skip += offset + return z.dpos + z.skip, nil +} + +// Reset discards the Reader's state and makes it equivalent to the +// result of its original state from NewReader, but reading from r instead. +// This permits reusing a Reader rather than allocating a new one. +func (z *ReaderLegacy) Reset(r io.Reader) { + z.Header = Header{} + z.pos = 0 + z.src = r + z.zdata = z.zdata[:0] + z.data = z.data[:0] + z.idx = 0 +} + +// readUint32 reads an uint32 into the supplied buffer. +// The idea is to make use of the already allocated buffers avoiding additional allocations. +func (z *ReaderLegacy) readUint32() (uint32, error) { + buf := z.buf[:4] + _, err := io.ReadFull(z.src, buf) + x := binary.LittleEndian.Uint32(buf) + return x, err +} diff --git a/vendor/github.com/pierrec/lz4/writer.go b/vendor/github.com/pierrec/lz4/writer.go index 0120438025..f066d56305 100644 --- a/vendor/github.com/pierrec/lz4/writer.go +++ b/vendor/github.com/pierrec/lz4/writer.go @@ -4,21 +4,35 @@ import ( "encoding/binary" "fmt" "io" + "runtime" "github.com/pierrec/lz4/internal/xxh32" ) +// zResult contains the results of compressing a block. +type zResult struct { + size uint32 // Block header + data []byte // Compressed data + checksum uint32 // Data checksum +} + // Writer implements the LZ4 frame encoder. type Writer struct { Header + // Handler called when a block has been successfully written out. + // It provides the number of bytes written. + OnBlockDone func(size int) buf [19]byte // magic number(4) + header(flags(2)+[Size(8)+DictID(4)]+checksum(1)) does not exceed 19 bytes dst io.Writer // Destination. checksum xxh32.XXHZero // Frame checksum. - zdata []byte // Compressed data. - data []byte // Data to be compressed. + data []byte // Data to be compressed + buffer for compressed data. idx int // Index into data. hashtable [winSize]int // Hash table used in CompressBlock(). + + // For concurrency. + c chan chan zResult // Channel for block compression goroutines and writer goroutine. + err error // Any error encountered while writing to the underlying destination. } // NewWriter returns a new LZ4 frame encoder. @@ -26,28 +40,90 @@ type Writer struct { // The supplied Header is checked at the first Write. // It is ok to change it before the first Write but then not until a Reset() is performed. func NewWriter(dst io.Writer) *Writer { - return &Writer{dst: dst} + z := new(Writer) + z.Reset(dst) + return z +} + +// WithConcurrency sets the number of concurrent go routines used for compression. +// A negative value sets the concurrency to GOMAXPROCS. +func (z *Writer) WithConcurrency(n int) *Writer { + switch { + case n == 0 || n == 1: + z.c = nil + return z + case n < 0: + n = runtime.GOMAXPROCS(0) + } + z.c = make(chan chan zResult, n) + // Writer goroutine managing concurrent block compression goroutines. + go func() { + // Process next block compression item. + for c := range z.c { + // Read the next compressed block result. + // Waiting here ensures that the blocks are output in the order they were sent. + // The incoming channel is always closed as it indicates to the caller that + // the block has been processed. + res := <-c + n := len(res.data) + if n == 0 { + // Notify the block compression routine that we are done with its result. + // This is used when a sentinel block is sent to terminate the compression. + close(c) + return + } + // Write the block. + if err := z.writeUint32(res.size); err != nil && z.err == nil { + z.err = err + } + if _, err := z.dst.Write(res.data); err != nil && z.err == nil { + z.err = err + } + if z.BlockChecksum { + if err := z.writeUint32(res.checksum); err != nil && z.err == nil { + z.err = err + } + } + // It is now safe to release the buffer as no longer in use by any goroutine. + putBuffer(cap(res.data), res.data) + if h := z.OnBlockDone; h != nil { + h(n) + } + close(c) + } + }() + return z +} + +// newBuffers instantiates new buffers which size matches the one in Header. +// The returned buffers are for decompression and compression respectively. +func (z *Writer) newBuffers() { + bSize := z.Header.BlockMaxSize + buf := getBuffer(bSize) + z.data = buf[:bSize] // Uncompressed buffer is the first half. +} + +// freeBuffers puts the writer's buffers back to the pool. +func (z *Writer) freeBuffers() { + // Put the buffer back into the pool, if any. + putBuffer(z.Header.BlockMaxSize, z.data) + z.data = nil } // writeHeader builds and writes the header (magic+header) to the underlying io.Writer. func (z *Writer) writeHeader() error { // Default to 4Mb if BlockMaxSize is not set. if z.Header.BlockMaxSize == 0 { - z.Header.BlockMaxSize = bsMapID[7] + z.Header.BlockMaxSize = blockSize4M } // The only option that needs to be validated. bSize := z.Header.BlockMaxSize - bSizeID, ok := bsMapValue[bSize] - if !ok { + if !isValidBlockSize(z.Header.BlockMaxSize) { return fmt.Errorf("lz4: invalid block max size: %d", bSize) } // Allocate the compressed/uncompressed buffers. // The compressed buffer cannot exceed the uncompressed one. - if n := 2 * bSize; cap(z.zdata) < n { - z.zdata = make([]byte, n, n) - } - z.zdata = z.zdata[:bSize] - z.data = z.zdata[:cap(z.zdata)][bSize:] + z.newBuffers() z.idx = 0 // Size is optional. @@ -67,7 +143,7 @@ func (z *Writer) writeHeader() error { flg |= 1 << 2 } buf[4] = flg - buf[5] = bSizeID << 4 + buf[5] = blockSizeValueToIndex(z.Header.BlockMaxSize) << 4 // Current buffer size: magic(4) + flags(1) + block max size (1). n := 6 @@ -147,28 +223,39 @@ func (z *Writer) Write(buf []byte) (int, error) { // compressBlock compresses a block. func (z *Writer) compressBlock(data []byte) error { if !z.NoChecksum { - z.checksum.Write(data) + _, _ = z.checksum.Write(data) } + if z.c != nil { + c := make(chan zResult) + z.c <- c // Send now to guarantee order + + // get a buffer from the pool and copy the data over + block := getBuffer(z.Header.BlockMaxSize)[:len(data)] + copy(block, data) + + go writerCompressBlock(c, z.Header, block) + return nil + } + + zdata := z.data[z.Header.BlockMaxSize:cap(z.data)] // The compressed block size cannot exceed the input's. var zn int - var err error if level := z.Header.CompressionLevel; level != 0 { - zn, err = CompressBlockHC(data, z.zdata, level) + zn, _ = CompressBlockHC(data, zdata, level) } else { - zn, err = CompressBlock(data, z.zdata, z.hashtable[:]) + zn, _ = CompressBlock(data, zdata, z.hashtable[:]) } - var zdata []byte var bLen uint32 if debugFlag { debug("block compression %d => %d", len(data), zn) } - if err == nil && zn > 0 && zn < len(data) { + if zn > 0 && zn < len(data) { // Compressible and compressed size smaller than uncompressed: ok! bLen = uint32(zn) - zdata = z.zdata[:zn] + zdata = zdata[:zn] } else { // Uncompressed block. bLen = uint32(len(data)) | compressedBlockFlag @@ -182,24 +269,26 @@ func (z *Writer) compressBlock(data []byte) error { if err := z.writeUint32(bLen); err != nil { return err } - if _, err := z.dst.Write(zdata); err != nil { + written, err := z.dst.Write(zdata) + if err != nil { return err } + if h := z.OnBlockDone; h != nil { + h(written) + } - if z.BlockChecksum { - checksum := xxh32.ChecksumZero(zdata) + if !z.BlockChecksum { if debugFlag { - debug("block checksum %x", checksum) - } - if err := z.writeUint32(checksum); err != nil { - return err + debug("current frame checksum %x", z.checksum.Sum32()) } + return nil } + checksum := xxh32.ChecksumZero(zdata) if debugFlag { - debug("current frame checksum %x", z.checksum.Sum32()) + debug("block checksum %x", checksum) + defer func() { debug("current frame checksum %x", z.checksum.Sum32()) }() } - - return nil + return z.writeUint32(checksum) } // Flush flushes any pending compressed data to the underlying writer. @@ -213,7 +302,35 @@ func (z *Writer) Flush() error { return nil } - return z.compressBlock(z.data[:z.idx]) + data := getBuffer(z.Header.BlockMaxSize)[:len(z.data[:z.idx])] + copy(data, z.data[:z.idx]) + + z.idx = 0 + if z.c == nil { + return z.compressBlock(data) + } + if !z.NoChecksum { + _, _ = z.checksum.Write(data) + } + c := make(chan zResult) + z.c <- c + writerCompressBlock(c, z.Header, data) + return nil +} + +func (z *Writer) close() error { + if z.c == nil { + return nil + } + // Send a sentinel block (no data to compress) to terminate the writer main goroutine. + c := make(chan zResult) + z.c <- c + c <- zResult{} + // Wait for the main goroutine to complete. + <-c + // At this point the main goroutine has shut down or is about to return. + z.c = nil + return z.err } // Close closes the Writer, flushing any unwritten data to the underlying io.Writer, but does not close the underlying io.Writer. @@ -223,10 +340,13 @@ func (z *Writer) Close() error { return err } } - if err := z.Flush(); err != nil { return err } + if err := z.close(); err != nil { + return err + } + z.freeBuffers() if debugFlag { debug("writing last empty block") @@ -234,28 +354,33 @@ func (z *Writer) Close() error { if err := z.writeUint32(0); err != nil { return err } - if !z.NoChecksum { - checksum := z.checksum.Sum32() - if debugFlag { - debug("stream checksum %x", checksum) - } - if err := z.writeUint32(checksum); err != nil { - return err - } + if z.NoChecksum { + return nil } - return nil + checksum := z.checksum.Sum32() + if debugFlag { + debug("stream checksum %x", checksum) + } + return z.writeUint32(checksum) } // Reset clears the state of the Writer z such that it is equivalent to its // initial state from NewWriter, but instead writing to w. // No access to the underlying io.Writer is performed. func (z *Writer) Reset(w io.Writer) { - z.Header = Header{} + n := cap(z.c) + _ = z.close() + z.freeBuffers() + z.Header.Reset() z.dst = w z.checksum.Reset() - z.zdata = z.zdata[:0] - z.data = z.data[:0] z.idx = 0 + z.err = nil + // reset hashtable to ensure deterministic output. + for i := range z.hashtable { + z.hashtable[i] = 0 + } + z.WithConcurrency(n) } // writeUint32 writes a uint32 to the underlying writer. @@ -265,3 +390,33 @@ func (z *Writer) writeUint32(x uint32) error { _, err := z.dst.Write(buf) return err } + +// writerCompressBlock compresses data into a pooled buffer and writes its result +// out to the input channel. +func writerCompressBlock(c chan zResult, header Header, data []byte) { + zdata := getBuffer(header.BlockMaxSize) + // The compressed block size cannot exceed the input's. + var zn int + if level := header.CompressionLevel; level != 0 { + zn, _ = CompressBlockHC(data, zdata, level) + } else { + var hashTable [winSize]int + zn, _ = CompressBlock(data, zdata, hashTable[:]) + } + var res zResult + if zn > 0 && zn < len(data) { + res.size = uint32(zn) + res.data = zdata[:zn] + // release the uncompressed block since it is not used anymore + putBuffer(header.BlockMaxSize, data) + } else { + res.size = uint32(len(data)) | compressedBlockFlag + res.data = data + // release the compressed block since it was not used + putBuffer(header.BlockMaxSize, zdata) + } + if header.BlockChecksum { + res.checksum = xxh32.ChecksumZero(res.data) + } + c <- res +} diff --git a/vendor/github.com/pierrec/lz4/writer_legacy.go b/vendor/github.com/pierrec/lz4/writer_legacy.go new file mode 100644 index 0000000000..ca8dc8c7f0 --- /dev/null +++ b/vendor/github.com/pierrec/lz4/writer_legacy.go @@ -0,0 +1,182 @@ +package lz4 + +import ( + "encoding/binary" + "io" +) + +// WriterLegacy implements the LZ4Demo frame decoder. +type WriterLegacy struct { + Header + // Handler called when a block has been successfully read. + // It provides the number of bytes read. + OnBlockDone func(size int) + + dst io.Writer // Destination. + data []byte // Data to be compressed + buffer for compressed data. + idx int // Index into data. + hashtable [winSize]int // Hash table used in CompressBlock(). +} + +// NewWriterLegacy returns a new LZ4 encoder for the legacy frame format. +// No access to the underlying io.Writer is performed. +// The supplied Header is checked at the first Write. +// It is ok to change it before the first Write but then not until a Reset() is performed. +func NewWriterLegacy(dst io.Writer) *WriterLegacy { + z := new(WriterLegacy) + z.Reset(dst) + return z +} + +// Write compresses data from the supplied buffer into the underlying io.Writer. +// Write does not return until the data has been written. +func (z *WriterLegacy) Write(buf []byte) (int, error) { + if !z.Header.done { + if err := z.writeHeader(); err != nil { + return 0, err + } + } + if debugFlag { + debug("input buffer len=%d index=%d", len(buf), z.idx) + } + + zn := len(z.data) + var n int + for len(buf) > 0 { + if z.idx == 0 && len(buf) >= zn { + // Avoid a copy as there is enough data for a block. + if err := z.compressBlock(buf[:zn]); err != nil { + return n, err + } + n += zn + buf = buf[zn:] + continue + } + // Accumulate the data to be compressed. + m := copy(z.data[z.idx:], buf) + n += m + z.idx += m + buf = buf[m:] + if debugFlag { + debug("%d bytes copied to buf, current index %d", n, z.idx) + } + + if z.idx < len(z.data) { + // Buffer not filled. + if debugFlag { + debug("need more data for compression") + } + return n, nil + } + + // Buffer full. + if err := z.compressBlock(z.data); err != nil { + return n, err + } + z.idx = 0 + } + + return n, nil +} + +// writeHeader builds and writes the header to the underlying io.Writer. +func (z *WriterLegacy) writeHeader() error { + // Legacy has fixed 8MB blocksizes + // https://github.com/lz4/lz4/blob/dev/doc/lz4_Frame_format.md#legacy-frame + bSize := 2 * blockSize4M + + buf := make([]byte, 2*bSize, 2*bSize) + z.data = buf[:bSize] // Uncompressed buffer is the first half. + + z.idx = 0 + + // Header consists of one mageic number, write it out. + if err := binary.Write(z.dst, binary.LittleEndian, frameMagicLegacy); err != nil { + return err + } + z.Header.done = true + if debugFlag { + debug("wrote header %v", z.Header) + } + + return nil +} + +// compressBlock compresses a block. +func (z *WriterLegacy) compressBlock(data []byte) error { + bSize := 2 * blockSize4M + zdata := z.data[bSize:cap(z.data)] + // The compressed block size cannot exceed the input's. + var zn int + + if level := z.Header.CompressionLevel; level != 0 { + zn, _ = CompressBlockHC(data, zdata, level) + } else { + zn, _ = CompressBlock(data, zdata, z.hashtable[:]) + } + + if debugFlag { + debug("block compression %d => %d", len(data), zn) + } + zdata = zdata[:zn] + + // Write the block. + if err := binary.Write(z.dst, binary.LittleEndian, uint32(zn)); err != nil { + return err + } + written, err := z.dst.Write(zdata) + if err != nil { + return err + } + if h := z.OnBlockDone; h != nil { + h(written) + } + return nil +} + +// Flush flushes any pending compressed data to the underlying writer. +// Flush does not return until the data has been written. +// If the underlying writer returns an error, Flush returns that error. +func (z *WriterLegacy) Flush() error { + if debugFlag { + debug("flush with index %d", z.idx) + } + if z.idx == 0 { + return nil + } + + data := z.data[:z.idx] + z.idx = 0 + return z.compressBlock(data) +} + +// Close closes the WriterLegacy, flushing any unwritten data to the underlying io.Writer, but does not close the underlying io.Writer. +func (z *WriterLegacy) Close() error { + if !z.Header.done { + if err := z.writeHeader(); err != nil { + return err + } + } + if err := z.Flush(); err != nil { + return err + } + + if debugFlag { + debug("writing last empty block") + } + + return nil +} + +// Reset clears the state of the WriterLegacy z such that it is equivalent to its +// initial state from NewWriterLegacy, but instead writing to w. +// No access to the underlying io.Writer is performed. +func (z *WriterLegacy) Reset(w io.Writer) { + z.Header.Reset() + z.dst = w + z.idx = 0 + // reset hashtable to ensure deterministic output. + for i := range z.hashtable { + z.hashtable[i] = 0 + } +} diff --git a/vendor/modules.txt b/vendor/modules.txt index eb35b4aeec..81d0640e77 100644 --- a/vendor/modules.txt +++ b/vendor/modules.txt @@ -587,7 +587,7 @@ github.com/openzipkin/zipkin-go-opentracing/types github.com/openzipkin/zipkin-go-opentracing/wire # github.com/philhofer/fwd v1.0.0 github.com/philhofer/fwd -# github.com/pierrec/lz4 v2.0.5+incompatible +# github.com/pierrec/lz4 v2.6.1+incompatible github.com/pierrec/lz4 github.com/pierrec/lz4/internal/xxh32 # github.com/pierrec/lz4/v4 v4.1.12 @@ -1226,7 +1226,7 @@ yunion.io/x/pkg/util/workqueue yunion.io/x/pkg/utils # yunion.io/x/s3cli v0.0.0-20190917004522-13ac36d8687e yunion.io/x/s3cli -# yunion.io/x/sqlchemy v1.1.1 +# yunion.io/x/sqlchemy v1.1.2-0.20220530125156-f4b471d64c78 yunion.io/x/sqlchemy yunion.io/x/sqlchemy/backends yunion.io/x/sqlchemy/backends/clickhouse diff --git a/vendor/yunion.io/x/sqlchemy/backends.go b/vendor/yunion.io/x/sqlchemy/backends.go index 2e3d3c5089..ce8b5955c7 100644 --- a/vendor/yunion.io/x/sqlchemy/backends.go +++ b/vendor/yunion.io/x/sqlchemy/backends.go @@ -50,7 +50,9 @@ type IBackend interface { // GetColumnSpecByFieldType parse the field of model struct to extract column specifiction of a field GetColumnSpecByFieldType(table *STableSpec, fieldType reflect.Type, fieldname string, tagmap map[string]string, isPointer bool) IColumnSpec // - // CurrentUTCTimeStampString() string + CurrentUTCTimeStampString() string + // + CaseInsensitiveLikeString() string // Capability @@ -74,6 +76,45 @@ type IBackend interface { // CommitTableChangeSQL outputs the SQLs to alter a table CommitTableChangeSQL(ts ITableSpec, changes STableChanges) []string + + /////////////////////////////////////////////////////////////////////// + ////////////////// FUNCTIONS ////////////////////////////////////////// + /////////////////////////////////////////////////////////////////////// + + // cast + CAST(field IQueryField, typeStr string, fieldname string) IQueryField + // TIMESTAMPADD + TIMESTAMPADD(name string, field IQueryField, offsetSeconds int) IQueryField + // INET_ATON + INET_ATON(field IQueryField) IQueryField + // AND_Val + AND_Val(name string, field IQueryField, v interface{}) IQueryField + // OR_Val + OR_Val(name string, field IQueryField, v interface{}) IQueryField + // SUBSTR + SUBSTR(name string, field IQueryField, pos, length int) IQueryField + // CONCAT + CONCAT(name string, fields ...IQueryField) IQueryField + // REPLACE + REPLACE(name string, field IQueryField, old string, new string) IQueryField + // GROUP_CONCAT2 + GROUP_CONCAT2(name string, sep string, field IQueryField) IQueryField + // DISTINCT + DISTINCT(name string, field IQueryField) IQueryField + // COUNT + COUNT(name string, field ...IQueryField) IQueryField + // MAX + MAX(name string, field IQueryField) IQueryField + // MIN + MIN(name string, field IQueryField) IQueryField + // SUM + SUM(name string, field IQueryField) IQueryField + // LENGTH + LENGTH(name string, field IQueryField) IQueryField + // LOWER + LOWER(name string, field IQueryField) IQueryField + // UPPER + UPPER(name string, field IQueryField) IQueryField } var _driver_tbl = make(map[DBBackendName]IBackend) diff --git a/vendor/yunion.io/x/sqlchemy/backends/clickhouse/clickhouse.go b/vendor/yunion.io/x/sqlchemy/backends/clickhouse/clickhouse.go index 46c96cf6ae..5377eca1f1 100644 --- a/vendor/yunion.io/x/sqlchemy/backends/clickhouse/clickhouse.go +++ b/vendor/yunion.io/x/sqlchemy/backends/clickhouse/clickhouse.go @@ -41,6 +41,10 @@ func (click *SClickhouseBackend) Name() sqlchemy.DBBackendName { return sqlchemy.ClickhouseBackend } +func (click *SClickhouseBackend) CaseInsensitiveLikeString() string { + return "ILIKE" +} + // CanUpdate returns wether the backend supports update func (click *SClickhouseBackend) CanUpdate() bool { return false diff --git a/vendor/yunion.io/x/sqlchemy/backends/clickhouse/functions.go b/vendor/yunion.io/x/sqlchemy/backends/clickhouse/functions.go new file mode 100644 index 0000000000..8be83ea882 --- /dev/null +++ b/vendor/yunion.io/x/sqlchemy/backends/clickhouse/functions.go @@ -0,0 +1,26 @@ +// Copyright 2019 Yunion +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package clickhouse + +import ( + "fmt" + + "yunion.io/x/sqlchemy" +) + +// GROUP_CONCAT1 represents the SQL function GROUP_CONCAT +func (click *SClickhouseBackend) GROUP_CONCAT1(name string, sep string, field sqlchemy.IQueryField) sqlchemy.IQueryField { + return sqlchemy.NewFunctionField(name, fmt.Sprintf("arrayStringConcat(groupUniqArray(%%s), '%s')", sep), field) +} diff --git a/vendor/yunion.io/x/sqlchemy/backends/mysql/functions.go b/vendor/yunion.io/x/sqlchemy/backends/mysql/functions.go new file mode 100644 index 0000000000..59a6b1cf6e --- /dev/null +++ b/vendor/yunion.io/x/sqlchemy/backends/mysql/functions.go @@ -0,0 +1,26 @@ +// Copyright 2019 Yunion +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package mysql + +import ( + "fmt" + + "yunion.io/x/sqlchemy" +) + +// GROUP_CONCAT2 represents the SQL function GROUP_CONCAT +func (mysql *SMySQLBackend) GROUP_CONCAT2(name string, sep string, field sqlchemy.IQueryField) sqlchemy.IQueryField { + return sqlchemy.NewFunctionField(name, fmt.Sprintf("GROUP_CONCAT(%%s SEPARATOR '%s')", sep), field) +} diff --git a/vendor/yunion.io/x/sqlchemy/backends/sqlite/functions.go b/vendor/yunion.io/x/sqlchemy/backends/sqlite/functions.go new file mode 100644 index 0000000000..a374f471a7 --- /dev/null +++ b/vendor/yunion.io/x/sqlchemy/backends/sqlite/functions.go @@ -0,0 +1,26 @@ +// Copyright 2019 Yunion +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package sqlite + +import ( + "fmt" + + "yunion.io/x/sqlchemy" +) + +// GROUP_CONCAT2 represents the SQL function GROUP_CONCAT +func (sqlite *SSqliteBackend) GROUP_CONCAT2(name string, sep string, field sqlchemy.IQueryField) sqlchemy.IQueryField { + return sqlchemy.NewFunctionField(name, fmt.Sprintf("GROUP_CONCAT(%%s, '%s')", sep), field) +} diff --git a/vendor/yunion.io/x/sqlchemy/backends_base.go b/vendor/yunion.io/x/sqlchemy/backends_base.go index 86ecf564bb..ff4e493138 100644 --- a/vendor/yunion.io/x/sqlchemy/backends_base.go +++ b/vendor/yunion.io/x/sqlchemy/backends_base.go @@ -14,13 +14,64 @@ package sqlchemy -type SBaseBackend struct { +import ( + "fmt" + "reflect" + "strings" +) + +var defaultBackend IBackend = (*SBaseBackend)(nil) + +type SBaseBackend struct{} + +func (bb *SBaseBackend) Name() DBBackendName { + return "default" +} + +func (bb *SBaseBackend) GetCreateSQLs(ts ITableSpec) []string { + return []string{} +} + +func (bb *SBaseBackend) IsSupportIndexAndContraints() bool { + return false } func (bb *SBaseBackend) GetTableSQL() string { return "SHOW TABLES" } +func (bb *SBaseBackend) FetchTableColumnSpecs(ts ITableSpec) ([]IColumnSpec, error) { + return nil, nil +} + +func (bb *SBaseBackend) GetColumnSpecByFieldType(table *STableSpec, fieldType reflect.Type, fieldname string, tagmap map[string]string, isPointer bool) IColumnSpec { + return nil +} + +func (bb *SBaseBackend) CurrentUTCTimeStampString() string { + return "NOW()" +} + +func (bb *SBaseBackend) CaseInsensitiveLikeString() string { + return "LIKE" +} + +func (bb *SBaseBackend) CanUpdate() bool { + return false +} + +func (bb *SBaseBackend) CanInsert() bool { + return false +} + +func (bb *SBaseBackend) CanInsertOrUpdate() bool { + return false +} + +func (bb *SBaseBackend) CommitTableChangeSQL(ts ITableSpec, changes STableChanges) []string { + return nil +} + func (bb *SBaseBackend) FetchIndexesAndConstraints(ts ITableSpec) ([]STableIndex, []STableConstraint, error) { return nil, nil, nil } @@ -40,3 +91,104 @@ func (bb *SBaseBackend) InsertSQLTemplate() string { func (bb *SBaseBackend) InsertOrUpdateSQLTemplate() string { return "" } + +func (bb *SBaseBackend) CAST(field IQueryField, typeStr string, fieldname string) IQueryField { + return NewFunctionField(fieldname, `CAST(%s AS `+typeStr+`)`, field) +} + +// TimestampAdd represents a SQL function TimestampAdd +func (bb *SBaseBackend) TIMESTAMPADD(name string, field IQueryField, offsetSeconds int) IQueryField { + return NewFunctionField(name, `TIMESTAMPADD(SECOND, `+fmt.Sprintf("%d", offsetSeconds)+`, %s)`, field) +} + +// INET_ATON represents a SQL function INET_ATON +func (bb *SBaseBackend) INET_ATON(field IQueryField) IQueryField { + return NewFunctionField("", `INET_ATON(%s)`, field) +} + +// SubStr represents a SQL function SUBSTR +func (bb *SBaseBackend) SUBSTR(name string, field IQueryField, pos, length int) IQueryField { + var rightStr string + if length <= 0 { + rightStr = fmt.Sprintf("%d)", pos) + } else { + rightStr = fmt.Sprintf("%d, %d)", pos, length) + } + return NewFunctionField(name, `SUBSTR(%s, `+rightStr, field) +} + +// OR_Val represents a SQL function that does binary | operation on a field +func (bb *SBaseBackend) OR_Val(name string, field IQueryField, v interface{}) IQueryField { + rightStr := fmt.Sprintf("|%v", v) + return NewFunctionField(name, "%s"+rightStr, field) +} + +// AND_Val represents a SQL function that does binary & operation on a field +func (bb *SBaseBackend) AND_Val(name string, field IQueryField, v interface{}) IQueryField { + rightStr := fmt.Sprintf("&%v", v) + return NewFunctionField(name, "%s"+rightStr, field) +} + +// CONCAT represents a SQL function CONCAT +func (bb *SBaseBackend) CONCAT(name string, fields ...IQueryField) IQueryField { + params := []string{} + for i := 0; i < len(fields); i++ { + params = append(params, "%s") + } + return NewFunctionField(name, `CONCAT(`+strings.Join(params, ",")+`)`, fields...) +} + +// REPLACE represents a SQL function REPLACE +func (bb *SBaseBackend) REPLACE(name string, field IQueryField, old string, new string) IQueryField { + return NewFunctionField(name, fmt.Sprintf(`REPLACE(%s, "%s", "%s")`, "%s", old, new), field) +} + +// DISTINCT represents the SQL function DISTINCT +func (bb *SBaseBackend) DISTINCT(name string, field IQueryField) IQueryField { + return NewFunctionField(name, "DISTINCT(%s)", field) +} + +// COUNT represents the SQL function COUNT +func (bb *SBaseBackend) COUNT(name string, field ...IQueryField) IQueryField { + var expr string + if len(field) == 0 { + expr = "COUNT(*)" + } else { + expr = "COUNT(%s)" + } + return NewFunctionField(name, expr, field...) +} + +// MAX represents the SQL function MAX +func (bb *SBaseBackend) MAX(name string, field IQueryField) IQueryField { + return NewFunctionField(name, "MAX(%s)", field) +} + +// MIN represents the SQL function MIN +func (bb *SBaseBackend) MIN(name string, field IQueryField) IQueryField { + return NewFunctionField(name, "MIN(%s)", field) +} + +// SUM represents the SQL function SUM +func (bb *SBaseBackend) SUM(name string, field IQueryField) IQueryField { + return NewFunctionField(name, "SUM(%s)", field) +} + +// LENGTH represents SQL function LENGTH +func (bb *SBaseBackend) LENGTH(name string, field IQueryField) IQueryField { + return NewFunctionField(name, "LENGTH(%s)", field) +} + +func (bb *SBaseBackend) GROUP_CONCAT2(name string, sep string, field IQueryField) IQueryField { + return NewFunctionField(name, fmt.Sprintf("GROUP_CONCAT(%%s SEPARATOR '%s')", sep), field) +} + +// LOWER represents SQL function of LOWER +func (bb *SBaseBackend) LOWER(name string, field IQueryField) IQueryField { + return NewFunctionField(name, "LOWER(%s)", field) +} + +// UPPER represents SQL function of UPPER +func (bb *SBaseBackend) UPPER(name string, field IQueryField) IQueryField { + return NewFunctionField(name, "UPPER(%s)", field) +} diff --git a/vendor/yunion.io/x/sqlchemy/case.go b/vendor/yunion.io/x/sqlchemy/case.go index 063d0adc8c..d63b3d87b4 100644 --- a/vendor/yunion.io/x/sqlchemy/case.go +++ b/vendor/yunion.io/x/sqlchemy/case.go @@ -29,15 +29,6 @@ type SCaseFunction struct { elseField IQueryField } -// NewFunction creates a field with SQL function -// for example: SUM(count) as total -func NewFunction(ifunc IFunction, name string) IQueryField { - return &SFunctionFieldBase{ - IFunction: ifunc, - alias: name, - } -} - // Else adds else clause for case when function func (cf *SCaseFunction) Else(field IQueryField) *SCaseFunction { cf.elseField = field @@ -85,3 +76,27 @@ func (cf *SCaseFunction) variables() []interface{} { vars = append(vars, fromvars...) return vars } + +func (cf *SCaseFunction) database() *SDatabase { + for _, b := range cf.branches { + db := b.whenCondition.database() + if db != nil { + return db + } + db = b.thenField.database() + if db != nil { + return db + } + } + db := cf.elseField.database() + if db != nil { + return db + } + return nil +} + +func (cf *SCaseFunction) queryFields() []IQueryField { + return []IQueryField{ + cf.elseField, + } +} diff --git a/vendor/yunion.io/x/sqlchemy/conditions.go b/vendor/yunion.io/x/sqlchemy/conditions.go index 4628af41ef..7c8a20ad29 100644 --- a/vendor/yunion.io/x/sqlchemy/conditions.go +++ b/vendor/yunion.io/x/sqlchemy/conditions.go @@ -27,6 +27,8 @@ import ( type ICondition interface { WhereClause() string Variables() []interface{} + + database() *SDatabase } // SCompoundConditions is a Compound condition represents AND or OR boolean operation @@ -67,6 +69,17 @@ func (c *SCompoundConditions) Variables() []interface{} { return vars } +// database implementation of SCompoundConditions for ICondition +func (c *SCompoundConditions) database() *SDatabase { + for _, c := range c.conditions { + db := c.database() + if db != nil { + return db + } + } + return nil +} + // SAndConditions represents the AND condition, which is a SCompoundConditions type SAndConditions struct { SCompoundConditions @@ -132,6 +145,11 @@ func (c *SNotCondition) Variables() []interface{} { return c.condition.Variables() } +// database implementation of SNotCondition for ICondition +func (c *SNotCondition) database() *SDatabase { + return c.condition.database() +} + // NOT method that makes negative operator on a condition func NOT(cond ICondition) ICondition { cc := SNotCondition{condition: cond} @@ -145,7 +163,12 @@ type SSingleCondition struct { // Variables implementation of SSingleCondition for ICondition func (c *SSingleCondition) Variables() []interface{} { - return []interface{}{} + return c.field.Variables() +} + +// database implementation of SSingleCondition for ICondition +func (c *SSingleCondition) database() *SDatabase { + return c.field.database() } // NewSingleCondition returns an instance of SSingleCondition @@ -192,12 +215,12 @@ type SIsEmptyCondition struct { // WhereClause implementation of SIsEmptyCondition for ICondition func (c *SIsEmptyCondition) WhereClause() string { - return fmt.Sprintf("LENGTH(%s) = 0", c.field.Reference()) + return fmt.Sprintf("%s = 0", c.field.Reference()) } // IsEmpty method that justifies where a text field is empty, e.g. length is zero func IsEmpty(f IQueryField) ICondition { - c := SIsEmptyCondition{NewSingleCondition(f)} + c := SIsEmptyCondition{NewSingleCondition(LENGTH("", f))} return &c } @@ -208,12 +231,13 @@ type SIsNullOrEmptyCondition struct { // WhereClause implementation of SIsNullOrEmptyCondition for ICondition func (c *SIsNullOrEmptyCondition) WhereClause() string { - return fmt.Sprintf("%s IS NULL OR LENGTH(%s) = 0", c.field.Reference(), c.field.Reference()) + originField := c.field.(*SFunctionFieldBase).queryFields()[0] + return fmt.Sprintf("%s IS NULL OR %s = 0", originField.Reference(), c.field.Reference()) } // IsNullOrEmpty is the ethod justifies a field is null or empty, e.g. a is null or length(a) == 0 func IsNullOrEmpty(f IQueryField) ICondition { - c := SIsNullOrEmptyCondition{NewSingleCondition(f)} + c := SIsNullOrEmptyCondition{NewSingleCondition(LENGTH("", f))} return &c } @@ -224,12 +248,13 @@ type SIsNotEmptyCondition struct { // WhereClause implementation of SIsNotEmptyCondition for ICondition func (c *SIsNotEmptyCondition) WhereClause() string { - return fmt.Sprintf("%s IS NOT NULL AND LENGTH(%s) > 0", c.field.Reference(), c.field.Reference()) + originField := c.field.(*SFunctionFieldBase).queryFields()[0] + return fmt.Sprintf("%s IS NOT NULL AND %s > 0", originField.Reference(), c.field.Reference()) } // IsNotEmpty method justifies a field is not empty func IsNotEmpty(f IQueryField) ICondition { - c := SIsNotEmptyCondition{NewSingleCondition(f)} + c := SIsNotEmptyCondition{NewSingleCondition(LENGTH("", f))} return &c } @@ -272,7 +297,8 @@ type SNoLaterThanCondition struct { // WhereClause implementation of SNoLaterThanCondition for ICondition func (c *SNoLaterThanCondition) WhereClause() string { - return fmt.Sprintf("%s <= NOW()", c.field.Reference()) + nowStr := c.field.database().backend.CurrentUTCTimeStampString() + return fmt.Sprintf("%s <= %s", c.field.Reference(), nowStr) } // NoLaterThan method justifies a DATETIME field is before current time @@ -288,7 +314,8 @@ type SNoEarlierThanCondition struct { // WhereClause implementation of SNoEarlierThanCondition for ICondition func (c *SNoEarlierThanCondition) WhereClause() string { - return fmt.Sprintf("%s >= NOW()", c.field.Reference()) + nowStr := c.field.database().backend.CurrentUTCTimeStampString() + return fmt.Sprintf("%s >= %s", c.field.Reference(), nowStr) } // NoEarlierThan justifies a field is no earlier than current time @@ -371,6 +398,11 @@ func (t *STupleCondition) Variables() []interface{} { return varConditionVariables(t.right) } +// database implementation of STupleCondition for ICondition +func (t *STupleCondition) database() *SDatabase { + return t.left.database() +} + // SInCondition represents a IN operation in SQL query type SInCondition struct { STupleCondition @@ -426,7 +458,8 @@ func likeEscape(s string) string { // WhereClause implementation for SLikeCondition for ICondition func (t *SLikeCondition) WhereClause() string { - return tupleConditionWhereClause(&t.STupleCondition, SQL_OP_LIKE) + op := t.left.database().backend.CaseInsensitiveLikeString() + return tupleConditionWhereClause(&t.STupleCondition, op) } // Like SQL operator @@ -613,6 +646,10 @@ func (t *STrueCondition) Variables() []interface{} { return nil } +func (t *STrueCondition) database() *SDatabase { + return nil +} + // SFalseCondition is a dummy condition that is always false type SFalseCondition struct{} @@ -625,3 +662,7 @@ func (t *SFalseCondition) WhereClause() string { func (t *SFalseCondition) Variables() []interface{} { return nil } + +func (t *SFalseCondition) database() *SDatabase { + return nil +} diff --git a/vendor/yunion.io/x/sqlchemy/errors.go b/vendor/yunion.io/x/sqlchemy/errors.go index b2250a55d9..dba208dd7e 100644 --- a/vendor/yunion.io/x/sqlchemy/errors.go +++ b/vendor/yunion.io/x/sqlchemy/errors.go @@ -53,5 +53,5 @@ const ( ErrUnionFieldsNotMatch = errors.Error("cannot union, name of fields not match") // ErrUnionDatabasesNotMatch is an Error constant: backend database of union queries not match - ErrUnionDatabasesNotMatch = errors.Error("cannot union across different databases") + ErrUnionAcrossDatabases = errors.Error("cannot union across different databases") ) diff --git a/vendor/yunion.io/x/sqlchemy/functions.go b/vendor/yunion.io/x/sqlchemy/functions.go index 2cb55ff7c1..269ee1323a 100644 --- a/vendor/yunion.io/x/sqlchemy/functions.go +++ b/vendor/yunion.io/x/sqlchemy/functions.go @@ -26,6 +26,17 @@ import ( type IFunction interface { expression() string variables() []interface{} + database() *SDatabase + queryFields() []IQueryField +} + +// NewFunction creates a field with SQL function +// for example: SUM(count) as total +func NewFunction(ifunc IFunction, name string) IQueryField { + return &SFunctionFieldBase{ + IFunction: ifunc, + alias: name, + } } // SFunctionFieldBase is a query field that is the result of a SQL embedded function, e.g. COUNT(*) as count @@ -37,7 +48,7 @@ type SFunctionFieldBase struct { // Reference implementation of SFunctionFieldBase for IQueryField func (ff *SFunctionFieldBase) Reference() string { if len(ff.alias) == 0 { - log.Warningf("reference a function field without alias! %s", ff.expression()) + // log.Warningf("reference a function field without alias! %s", ff.expression()) return ff.expression() } return fmt.Sprintf("`%s`", ff.alias) @@ -96,6 +107,21 @@ func (ff *sExprFunction) variables() []interface{} { return vars } +func (ff *sExprFunction) database() *SDatabase { + for i := range ff.fields { + db := ff.fields[i].database() + if db != nil { + return db + } + } + log.Debugf("no fields function? %s", ff.expression()) + return nil +} + +func (ff *sExprFunction) queryFields() []IQueryField { + return ff.fields +} + // NewFunctionField returns an instance of query field by calling a SQL embedded function func NewFunctionField(name string, funcexp string, fields ...IQueryField) IQueryField { funcBase := &sExprFunction{ @@ -110,43 +136,53 @@ func NewFunctionField(name string, funcexp string, fields ...IQueryField) IQuery // COUNT represents the SQL function COUNT func COUNT(name string, field ...IQueryField) IQueryField { - var expr string - if len(field) == 0 { - expr = "COUNT(*)" - } else { - expr = "COUNT(%s)" - } - return NewFunctionField(name, expr, field...) + return getFieldBackend(field...).COUNT(name, field...) } // MAX represents the SQL function MAX func MAX(name string, field IQueryField) IQueryField { - return NewFunctionField(name, "MAX(%s)", field) + return getFieldBackend(field).MAX(name, field) } // MIN represents the SQL function MIN func MIN(name string, field IQueryField) IQueryField { - return NewFunctionField(name, "MIN(%s)", field) + return getFieldBackend(field).MIN(name, field) } // SUM represents the SQL function SUM func SUM(name string, field IQueryField) IQueryField { - return NewFunctionField(name, "SUM(%s)", field) + return getFieldBackend(field).SUM(name, field) +} + +// LOWER represents the SQL function SUM +func LOWER(name string, field IQueryField) IQueryField { + return getFieldBackend(field).LOWER(name, field) +} + +// UPPER represents the SQL function SUM +func UPPER(name string, field IQueryField) IQueryField { + return getFieldBackend(field).UPPER(name, field) } // DISTINCT represents the SQL function DISTINCT func DISTINCT(name string, field IQueryField) IQueryField { - return NewFunctionField(name, "DISTINCT(%s)", field) + return getFieldBackend(field).DISTINCT(name, field) } // GROUP_CONCAT represents the SQL function GROUP_CONCAT func GROUP_CONCAT(name string, field IQueryField) IQueryField { - return NewFunctionField(name, "GROUP_CONCAT(%s)", field) + return GROUP_CONCAT2(name, ",", field) +} + +// GROUP_CONCAT2 represents the SQL function GROUP_CONCAT +func GROUP_CONCAT2(name string, sep string, field IQueryField) IQueryField { + // return NewFunctionField(name, "GROUP_CONCAT(%s)", field) + return getFieldBackend(field).GROUP_CONCAT2(name, sep, field) } // REPLACE represents the SQL function REPLACE func REPLACE(name string, field IQueryField, old string, new string) IQueryField { - return NewFunctionField(name, fmt.Sprintf(`REPLACE(%s, "%s", "%s")`, "%s", old, new), field) + return getFieldBackend(field).REPLACE(name, field, old, new) } // SConstField is a query field of a constant @@ -183,6 +219,11 @@ func (s *SConstField) Label(label string) IQueryField { return s } +// database implementation of SConstField for IQueryField +func (s *SConstField) database() *SDatabase { + return nil +} + // Variables implementation of SConstField for IQueryField func (s *SConstField) Variables() []interface{} { return nil @@ -227,23 +268,24 @@ func (s *SStringField) Label(label string) IQueryField { return s } +// database implementation of SStringField for IQueryField +func (s *SStringField) database() *SDatabase { + return nil +} + // Variables implementation of SStringField for IQueryField func (s *SStringField) Variables() []interface{} { return nil } // NewStringField returns an instance of SStringField -func NewStringField(name string) *SStringField { - return &SStringField{strConst: name} +func NewStringField(strConst string) *SStringField { + return &SStringField{strConst: strConst} } // CONCAT represents a SQL function CONCAT func CONCAT(name string, fields ...IQueryField) IQueryField { - params := []string{} - for i := 0; i < len(fields); i++ { - params = append(params, "%s") - } - return NewFunctionField(name, `CONCAT(`+strings.Join(params, ",")+`)`, fields...) + return getFieldBackend(fields...).CONCAT(name, fields...) } // SubStr represents a SQL function SUBSTR @@ -252,40 +294,44 @@ func SubStr(name string, field IQueryField, pos, length int) IQueryField { return SUBSTR(name, field, pos, length) } +// SUBSTR represents a SQL function SUBSTR func SUBSTR(name string, field IQueryField, pos, length int) IQueryField { - var rightStr string - if length <= 0 { - rightStr = fmt.Sprintf("%d)", pos) - } else { - rightStr = fmt.Sprintf("%d, %d)", pos, length) - } - return NewFunctionField(name, `SUBSTR(%s, `+rightStr, field) + return getFieldBackend(field).SUBSTR(name, field, pos, length) } // OR_Val represents a SQL function that does binary | operation on a field func OR_Val(name string, field IQueryField, v interface{}) IQueryField { - rightStr := fmt.Sprintf("|%v", v) - return NewFunctionField(name, "%s"+rightStr, field) + return getFieldBackend(field).OR_Val(name, field, v) } // AND_Val represents a SQL function that does binary & operation on a field func AND_Val(name string, field IQueryField, v interface{}) IQueryField { - rightStr := fmt.Sprintf("&%v", v) - return NewFunctionField(name, "%s"+rightStr, field) + return getFieldBackend(field).AND_Val(name, field, v) } // INET_ATON represents a SQL function INET_ATON func INET_ATON(field IQueryField) IQueryField { - return NewFunctionField("", `INET_ATON(%s)`, field) + return getFieldBackend(field).INET_ATON(field) } // TimestampAdd represents a SQL function TimestampAdd func TimestampAdd(name string, field IQueryField, offsetSeconds int) IQueryField { - return NewFunctionField(name, `TIMESTAMPADD(SECOND, `+fmt.Sprintf("%d", offsetSeconds)+`, %s)`, field) + return TIMESTAMPADD(name, field, offsetSeconds) } +// TIMESTAMPADD represents a SQL function TimestampAdd +func TIMESTAMPADD(name string, field IQueryField, offsetSeconds int) IQueryField { + return getFieldBackend(field).TIMESTAMPADD(name, field, offsetSeconds) +} + +// CAST represents a SQL function cast types func CAST(field IQueryField, typeStr string, fieldname string) IQueryField { - return NewFunctionField(fieldname, `CAST(%s AS `+typeStr+`)`, field) + return getFieldBackend(field).CAST(field, typeStr, fieldname) +} + +// LENGTH represents a SQL function of LENGTH +func LENGTH(name string, field IQueryField) IQueryField { + return getFieldBackend(field).LENGTH(name, field) } func bc(name, op string, fields ...IQueryField) IQueryField { diff --git a/vendor/yunion.io/x/sqlchemy/query.go b/vendor/yunion.io/x/sqlchemy/query.go index 2d5bb6c6a7..d9d9880b8d 100644 --- a/vendor/yunion.io/x/sqlchemy/query.go +++ b/vendor/yunion.io/x/sqlchemy/query.go @@ -15,11 +15,9 @@ package sqlchemy import ( - "bytes" "database/sql" "fmt" "reflect" - "sort" "strings" "yunion.io/x/log" @@ -27,83 +25,6 @@ import ( "yunion.io/x/pkg/util/reflectutils" ) -// IQuery is an interface that reprsents a SQL query, e.g. -// SELECT ... FROM ... WHERE ... -type IQuery interface { - // String returns the queryString - String(fields ...IQueryField) string - - // QueryFields returns fields in the select clause - QueryFields() []IQueryField - - // Variables returns variables in statement - Variables() []interface{} - - // SubQuery convert this SQL to a subquery - SubQuery() *SSubQuery - - // Field reference to a field by name - Field(name string) IQueryField - - // Database returns the database for this query - Database() *SDatabase -} - -// IQuerySource is an interface that represents a data source of a SQL query. the source can be a table or a subquery -// e.g. SELECT ... FROM (SELECT * FROM tbl) AS A -type IQuerySource interface { - // Expression string in select ... from (expresson here) - Expression() string - - // Alias is the alias in select ... from (express) as alias - Alias() string - - // variables in statement - Variables() []interface{} - - // Field reference to a field by name, optionally giving an alias name - Field(id string, alias ...string) IQueryField - - // Fields return all the fields that this source provides - Fields() []IQueryField - - // Database returns the database of this IQuerySource - Database() *SDatabase -} - -// IQueryField is an interface that represents a select field in a SQL query -type IQueryField interface { - // the string after select - Expression() string - - // the name of thie field - Name() string - - // the reference string in where clause - Reference() string - - // give this field an alias name - Label(label string) IQueryField - - // return variables - Variables() []interface{} -} - -// Expression implementation of STable for IQuerySource -func (tbl *STable) Expression() string { - return tbl.spec.Expression() -} - -// Alias implementation of STable for IQuerySource -func (tbl *STable) Alias() string { - return tbl.alias -} - -// Variables implementation of STable for IQuerySource -func (tbl *STable) Variables() []interface{} { - return []interface{}{} -} - // QueryJoinType is the Join type of SQL query, namely, innerjoin, leftjoin and rightjoin type QueryJoinType string @@ -154,137 +75,6 @@ func (tq *SQuery) IsGroupBy() bool { return len(tq.groupBy) > 0 } -// SSubQuery represents a subquery. A subquery is a query used as a query source -// SSubQuery should implementation IQuerySource -// At the same time, a subquery can be used in condition. e.g. IN condition -type SSubQuery struct { - query IQuery - alias string - - referedFields map[string]IQueryField -} - -// SSubQueryField represents a field of subquery, which implements IQueryField -type SSubQueryField struct { - field IQueryField - query *SSubQuery - alias string -} - -// Expression implementation of SSubQueryField for IQueryField -func (sqf *SSubQueryField) Expression() string { - if len(sqf.alias) > 0 { - return fmt.Sprintf("`%s`.`%s` AS `%s`", sqf.query.alias, sqf.field.Name(), sqf.alias) - } - return fmt.Sprintf("`%s`.`%s`", sqf.query.alias, sqf.field.Name()) -} - -// Name implementation of SSubQueryField for IQueryField -func (sqf *SSubQueryField) Name() string { - if len(sqf.alias) > 0 { - return sqf.alias - } - return sqf.field.Name() -} - -// Reference implementation of SSubQueryField for IQueryField -func (sqf *SSubQueryField) Reference() string { - return fmt.Sprintf("`%s`.`%s`", sqf.query.alias, sqf.Name()) -} - -// Label implementation of SSubQueryField for IQueryField -func (sqf *SSubQueryField) Label(label string) IQueryField { - if len(label) > 0 && label != sqf.field.Name() { - sqf.alias = label - } - return sqf -} - -// Variables implementation of SSubQueryField for IQueryField -func (sqf *SSubQueryField) Variables() []interface{} { - return nil -} - -// Expression implementation of SSubQuery for IQuerySource -func (sq *SSubQuery) Expression() string { - fields := make([]IQueryField, 0) - for k := range sq.referedFields { - fields = append(fields, sq.referedFields[k]) - } - // Make sure the order of the fields - sort.Slice(fields, func(i, j int) bool { - return fields[i].Name() < fields[j].Name() - }) - return fmt.Sprintf("(%s)", sq.query.String(fields...)) -} - -// Alias implementation of SSubQuery for IQuerySource -func (sq *SSubQuery) Alias() string { - return sq.alias -} - -// Variables implementation of SSubQuery for IQuerySource -func (sq *SSubQuery) Variables() []interface{} { - return sq.query.Variables() -} - -func (sq *SSubQuery) findField(id string) IQueryField { - if sq.referedFields == nil { - sq.referedFields = make(map[string]IQueryField) - } - if _, ok := sq.referedFields[id]; ok { - return sq.referedFields[id] - } - queryFields := sq.query.QueryFields() - for i := range queryFields { - if queryFields[i].Name() == id { - sq.referedFields[id] = sq.query.Field(queryFields[i].Name()) - return sq.referedFields[id] - } - } - return nil -} - -// Field implementation of SSubQuery for IQuerySource -func (sq *SSubQuery) Field(id string, alias ...string) IQueryField { - f := sq.findField(id) - if f == nil { - return nil - } - sqf := SSubQueryField{query: sq, field: f} - if len(alias) > 0 { - sqf.Label(alias[0]) - } - return &sqf -} - -// Fields implementation of SSubQuery for IQuerySource -func (sq *SSubQuery) Fields() []IQueryField { - ret := make([]IQueryField, 0) - for _, f := range sq.query.QueryFields() { - sqf := SSubQueryField{query: sq, field: f} - ret = append(ret, &sqf) - } - return ret -} - -// Database implementation of SSubQuery for IQuerySource -func (sq *SSubQuery) Database() *SDatabase { - return sq.query.Database() -} - -// DoQuery returns a SQuery instance that query specified fields from a query source -func DoQuery(from IQuerySource, f ...IQueryField) *SQuery { - if from.Database() == nil { - panic("DoQuery IQuerySource with empty database") - } - // if len(f) == 0 { - // f = from.Fields() - // } - tq := SQuery{fields: f, from: from, db: from.Database()} - return &tq -} - // AppendField appends query field to a query func (tq *SQuery) AppendField(f ...IQueryField) *SQuery { tq.fields = append(tq.fields, f...) @@ -403,77 +193,6 @@ func (tq *SQuery) String(fields ...IQueryField) string { return sql } -func queryString(tq *SQuery, tmpFields ...IQueryField) string { - if len(tq.rawSql) > 0 { - return tq.rawSql - } - - var buf bytes.Buffer - buf.WriteString("SELECT ") - if tq.distinct { - buf.WriteString("DISTINCT ") - } - fields := tq.fields - if len(fields) == 0 { - fields = tmpFields - } - if len(fields) == 0 { - fields = tq.QueryFields() - for i := range fields { - tq.from.Field(fields[i].Name()) - } - } - for i := range fields { - if i > 0 { - buf.WriteString(", ") - } - buf.WriteString(fields[i].Expression()) - } - buf.WriteString(" FROM ") - buf.WriteString(fmt.Sprintf("%s AS `%s`", tq.from.Expression(), tq.from.Alias())) - for _, join := range tq.joins { - buf.WriteByte(' ') - buf.WriteString(string(join.jointype)) - buf.WriteByte(' ') - buf.WriteString(fmt.Sprintf("%s AS `%s`", join.from.Expression(), join.from.Alias())) - buf.WriteString(" ON ") - buf.WriteString(join.condition.WhereClause()) - } - if tq.where != nil { - buf.WriteString(" WHERE ") - buf.WriteString(tq.where.WhereClause()) - } - if tq.groupBy != nil && len(tq.groupBy) > 0 { - buf.WriteString(" GROUP BY ") - for i, f := range tq.groupBy { - if i > 0 { - buf.WriteString(", ") - } - buf.WriteString(f.Reference()) - } - } - if tq.having != nil { - buf.WriteString(" HAVING ") - buf.WriteString(tq.having.WhereClause()) - } - if tq.orderBy != nil && len(tq.orderBy) > 0 { - buf.WriteString(" ORDER BY ") - for i, f := range tq.orderBy { - if i > 0 { - buf.WriteString(", ") - } - buf.WriteString(fmt.Sprintf("%s %s", f.field.Reference(), f.order)) - } - } - if tq.limit > 0 { - buf.WriteString(fmt.Sprintf(" LIMIT %d", tq.limit)) - } - if tq.offset > 0 { - buf.WriteString(fmt.Sprintf(" OFFSET %d", tq.offset)) - } - return buf.String() -} - // Join of SQuery joins query with another IQuerySource on specified condition func (tq *SQuery) Join(from IQuerySource, on ICondition) *SQuery { return tq._join(from, on, INNERJOIN) @@ -494,8 +213,8 @@ func (tq *SQuery) RightJoin(from IQuerySource, on ICondition) *SQuery { }*/ func (tq *SQuery) _join(from IQuerySource, on ICondition, joinType QueryJoinType) *SQuery { - if from.Database() != tq.db { - panic(fmt.Sprintf("Cannot join across databases %s!=%s", tq.db.name, from.Database().name)) + if from.database() != tq.db { + panic(fmt.Sprintf("Cannot join across databases %s!=%s", tq.db.name, from.database().name)) } if tq.joins == nil { tq.joins = make([]sQueryJoin, 0) @@ -547,7 +266,7 @@ func (tq *SQuery) SubQuery() *SSubQuery { return &sq } -func (tq *SQuery) Database() *SDatabase { +func (tq *SQuery) database() *SDatabase { return tq.db } @@ -594,7 +313,7 @@ func (tq *SQuery) CountQuery() *SQuery { COUNT("count"), }, from: tq2.SubQuery(), - db: tq.Database(), + db: tq.database(), } return cq } diff --git a/vendor/yunion.io/x/sqlchemy/querydefs.go b/vendor/yunion.io/x/sqlchemy/querydefs.go new file mode 100644 index 0000000000..6071d2b2fc --- /dev/null +++ b/vendor/yunion.io/x/sqlchemy/querydefs.go @@ -0,0 +1,178 @@ +// Copyright 2019 Yunion +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package sqlchemy + +import ( + "bytes" + "fmt" +) + +// IQuery is an interface that reprsents a SQL query, e.g. +// SELECT ... FROM ... WHERE ... +type IQuery interface { + // String returns the queryString + String(fields ...IQueryField) string + + // QueryFields returns fields in the select clause + QueryFields() []IQueryField + + // Variables returns variables in statement + Variables() []interface{} + + // SubQuery convert this SQL to a subquery + SubQuery() *SSubQuery + + // Field reference to a field by name + Field(name string) IQueryField + + // Database returns the database for this query + database() *SDatabase +} + +// IQuerySource is an interface that represents a data source of a SQL query. the source can be a table or a subquery +// e.g. SELECT ... FROM (SELECT * FROM tbl) AS A +type IQuerySource interface { + // Expression string in select ... from (expresson here) + Expression() string + + // Alias is the alias in select ... from (express) as alias + Alias() string + + // variables in statement + Variables() []interface{} + + // Field reference to a field by name, optionally giving an alias name + Field(id string, alias ...string) IQueryField + + // Fields return all the fields that this source provides + Fields() []IQueryField + + // Database returns the database of this IQuerySource + database() *SDatabase +} + +// IQueryField is an interface that represents a select field in a SQL query +type IQueryField interface { + // the string after select + Expression() string + + // the name of thie field + Name() string + + // the reference string in where clause + Reference() string + + // give this field an alias name + Label(label string) IQueryField + + // return variables + Variables() []interface{} + + // Database returns the database of this IQuerySource + database() *SDatabase +} + +// DoQuery returns a SQuery instance that query specified fields from a query source +func DoQuery(from IQuerySource, f ...IQueryField) *SQuery { + if from.database() == nil { + panic("DoQuery IQuerySource with empty database") + } + // if len(f) == 0 { + // f = from.Fields() + // } + tq := SQuery{fields: f, from: from, db: from.database()} + return &tq +} + +func queryString(tq *SQuery, tmpFields ...IQueryField) string { + if len(tq.rawSql) > 0 { + return tq.rawSql + } + + var buf bytes.Buffer + buf.WriteString("SELECT ") + if tq.distinct { + buf.WriteString("DISTINCT ") + } + fields := tq.fields + if len(fields) == 0 { + fields = tmpFields + } + if len(fields) == 0 { + fields = tq.QueryFields() + for i := range fields { + tq.from.Field(fields[i].Name()) + } + } + for i := range fields { + if i > 0 { + buf.WriteString(", ") + } + buf.WriteString(fields[i].Expression()) + } + buf.WriteString(" FROM ") + buf.WriteString(fmt.Sprintf("%s AS `%s`", tq.from.Expression(), tq.from.Alias())) + for _, join := range tq.joins { + buf.WriteByte(' ') + buf.WriteString(string(join.jointype)) + buf.WriteByte(' ') + buf.WriteString(fmt.Sprintf("%s AS `%s`", join.from.Expression(), join.from.Alias())) + buf.WriteString(" ON ") + buf.WriteString(join.condition.WhereClause()) + } + if tq.where != nil { + buf.WriteString(" WHERE ") + buf.WriteString(tq.where.WhereClause()) + } + if tq.groupBy != nil && len(tq.groupBy) > 0 { + buf.WriteString(" GROUP BY ") + for i, f := range tq.groupBy { + if i > 0 { + buf.WriteString(", ") + } + buf.WriteString(f.Reference()) + } + } + if tq.having != nil { + buf.WriteString(" HAVING ") + buf.WriteString(tq.having.WhereClause()) + } + if tq.orderBy != nil && len(tq.orderBy) > 0 { + buf.WriteString(" ORDER BY ") + for i, f := range tq.orderBy { + if i > 0 { + buf.WriteString(", ") + } + buf.WriteString(fmt.Sprintf("%s %s", f.field.Reference(), f.order)) + } + } + if tq.limit > 0 { + buf.WriteString(fmt.Sprintf(" LIMIT %d", tq.limit)) + } + if tq.offset > 0 { + buf.WriteString(fmt.Sprintf(" OFFSET %d", tq.offset)) + } + return buf.String() +} + +func getFieldBackend(fields ...IQueryField) IBackend { + for _, f := range fields { + db := f.database() + if db != nil { + return db.backend + } + } + return defaultBackend +} diff --git a/vendor/yunion.io/x/sqlchemy/rawquery.go b/vendor/yunion.io/x/sqlchemy/rawquery.go index cc575a8e67..854513c2ba 100644 --- a/vendor/yunion.io/x/sqlchemy/rawquery.go +++ b/vendor/yunion.io/x/sqlchemy/rawquery.go @@ -20,6 +20,7 @@ package sqlchemy // the struct implements IQueryField interface type SRawQueryField struct { name string + db *SDatabase } // Expression implementation of SRawQueryField for IQueryField @@ -47,6 +48,10 @@ func (rqf *SRawQueryField) Variables() []interface{} { return nil } +func (rqf *SRawQueryField) database() *SDatabase { + return rqf.db +} + // NewRawQuery returns an instance of SQuery with raw SQL query. e.g. show tables func NewRawQuery(sqlStr string, fields ...string) *SQuery { return GetDefaultDB().NewRawQuery(sqlStr, fields...) @@ -56,7 +61,10 @@ func NewRawQuery(sqlStr string, fields ...string) *SQuery { func (db *SDatabase) NewRawQuery(sqlStr string, fields ...string) *SQuery { qfs := make([]IQueryField, len(fields)) for i, f := range fields { - rqf := SRawQueryField{name: f} + rqf := SRawQueryField{ + name: f, + db: db, + } qfs[i] = &rqf } q := SQuery{ diff --git a/vendor/yunion.io/x/sqlchemy/subquery.go b/vendor/yunion.io/x/sqlchemy/subquery.go new file mode 100644 index 0000000000..831bcd68ec --- /dev/null +++ b/vendor/yunion.io/x/sqlchemy/subquery.go @@ -0,0 +1,144 @@ +// Copyright 2019 Yunion + +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package sqlchemy + +import ( + "fmt" + "sort" +) + +// SSubQueryField represents a field of subquery, which implements IQueryField +type SSubQueryField struct { + field IQueryField + query *SSubQuery + alias string +} + +// Expression implementation of SSubQueryField for IQueryField +func (sqf *SSubQueryField) Expression() string { + if len(sqf.alias) > 0 { + return fmt.Sprintf("`%s`.`%s` AS `%s`", sqf.query.alias, sqf.field.Name(), sqf.alias) + } + return fmt.Sprintf("`%s`.`%s`", sqf.query.alias, sqf.field.Name()) +} + +// Name implementation of SSubQueryField for IQueryField +func (sqf *SSubQueryField) Name() string { + if len(sqf.alias) > 0 { + return sqf.alias + } + return sqf.field.Name() +} + +// Reference implementation of SSubQueryField for IQueryField +func (sqf *SSubQueryField) Reference() string { + return fmt.Sprintf("`%s`.`%s`", sqf.query.alias, sqf.Name()) +} + +// Label implementation of SSubQueryField for IQueryField +func (sqf *SSubQueryField) Label(label string) IQueryField { + if len(label) > 0 && label != sqf.field.Name() { + sqf.alias = label + } + return sqf +} + +// Variables implementation of SSubQueryField for IQueryField +func (sqf *SSubQueryField) Variables() []interface{} { + return nil +} + +// database implementation of SSubQueryField for IQueryField +func (sqf *SSubQueryField) database() *SDatabase { + return sqf.query.database() +} + +// SSubQuery represents a subquery. A subquery is a query used as a query source +// SSubQuery should implementation IQuerySource +// At the same time, a subquery can be used in condition. e.g. IN condition +type SSubQuery struct { + query IQuery + alias string + + referedFields map[string]IQueryField +} + +// Expression implementation of SSubQuery for IQuerySource +func (sq *SSubQuery) Expression() string { + fields := make([]IQueryField, 0) + for k := range sq.referedFields { + fields = append(fields, sq.referedFields[k]) + } + // Make sure the order of the fields + sort.Slice(fields, func(i, j int) bool { + return fields[i].Name() < fields[j].Name() + }) + return fmt.Sprintf("(%s)", sq.query.String(fields...)) +} + +// Alias implementation of SSubQuery for IQuerySource +func (sq *SSubQuery) Alias() string { + return sq.alias +} + +// Variables implementation of SSubQuery for IQuerySource +func (sq *SSubQuery) Variables() []interface{} { + return sq.query.Variables() +} + +func (sq *SSubQuery) findField(id string) IQueryField { + if sq.referedFields == nil { + sq.referedFields = make(map[string]IQueryField) + } + if _, ok := sq.referedFields[id]; ok { + return sq.referedFields[id] + } + queryFields := sq.query.QueryFields() + for i := range queryFields { + if queryFields[i].Name() == id { + sq.referedFields[id] = sq.query.Field(queryFields[i].Name()) + return sq.referedFields[id] + } + } + return nil +} + +// Field implementation of SSubQuery for IQuerySource +func (sq *SSubQuery) Field(id string, alias ...string) IQueryField { + f := sq.findField(id) + if f == nil { + return nil + } + sqf := SSubQueryField{query: sq, field: f} + if len(alias) > 0 { + sqf.Label(alias[0]) + } + return &sqf +} + +// Fields implementation of SSubQuery for IQuerySource +func (sq *SSubQuery) Fields() []IQueryField { + ret := make([]IQueryField, 0) + for _, f := range sq.query.QueryFields() { + sqf := SSubQueryField{query: sq, field: f} + ret = append(ret, &sqf) + } + return ret +} + +// database implementation of SSubQuery for IQuerySource +func (sq *SSubQuery) database() *SDatabase { + return sq.query.database() +} diff --git a/vendor/yunion.io/x/sqlchemy/table.go b/vendor/yunion.io/x/sqlchemy/table.go index c81d4cf175..3888d07226 100644 --- a/vendor/yunion.io/x/sqlchemy/table.go +++ b/vendor/yunion.io/x/sqlchemy/table.go @@ -245,10 +245,25 @@ func (tbl *STable) Fields() []IQueryField { } // Database implementaion of STable for IQuerySource -func (tbl *STable) Database() *SDatabase { +func (tbl *STable) database() *SDatabase { return tbl.spec.Database() } +// Expression implementation of STable for IQuerySource +func (tbl *STable) Expression() string { + return tbl.spec.Expression() +} + +// Alias implementation of STable for IQuerySource +func (tbl *STable) Alias() string { + return tbl.alias +} + +// Variables implementation of STable for IQuerySource +func (tbl *STable) Variables() []interface{} { + return []interface{}{} +} + // Expression implementation of STableField for IQueryField func (c *STableField) Expression() string { if len(c.alias) > 0 { @@ -282,3 +297,8 @@ func (c *STableField) Label(label string) IQueryField { func (c *STableField) Variables() []interface{} { return nil } + +// database implementation of STableField for IQueryField +func (c *STableField) database() *SDatabase { + return c.table.database() +} diff --git a/vendor/yunion.io/x/sqlchemy/union.go b/vendor/yunion.io/x/sqlchemy/union.go index a930036d2c..f98dc3aeaa 100644 --- a/vendor/yunion.io/x/sqlchemy/union.go +++ b/vendor/yunion.io/x/sqlchemy/union.go @@ -64,6 +64,10 @@ func (sqf *SUnionQueryField) Variables() []interface{} { return nil } +func (sqf *SUnionQueryField) database() *SDatabase { + return sqf.union.database() +} + // SUnion is the struct to store state of a Union query, which implementation the interface of IQuerySource type SUnion struct { alias string @@ -85,7 +89,7 @@ func (uq *SUnion) operator() string { if uq.isAll { return " UNION ALL " } else { - return " UNION " + return " UNION DISTINCT " } } @@ -178,8 +182,14 @@ func (uq *SUnion) Variables() []interface{} { } // Database implementation of SUnion for IQUerySource -func (uq *SUnion) Database() *SDatabase { - return uq.queries[0].Database() +func (uq *SUnion) database() *SDatabase { + for _, q := range uq.queries { + db := q.database() + if db != nil { + return db + } + } + return nil } // Union method returns union query of several queries. @@ -216,9 +226,9 @@ func unionWithError(isAll bool, query ...IQuery) (*SUnion, error) { var db *SDatabase for i := 1; i < len(query); i++ { if db == nil { - db = query[i].Database() - } else if db != query[i].Database() { - panic(ErrUnionDatabasesNotMatch) + db = query[i].database() + } else if db != query[i].database() { + panic(ErrUnionAcrossDatabases) } qfields := query[i].QueryFields() if len(fieldNames) != len(qfields) {