Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
1 parent
66ca1f1
commit 31c4c0d
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,20 +1,26 @@ | ||
| package avro | ||
|
|
||
| import ( | ||
| "bytes" | ||
| "compress/flate" | ||
| "encoding/binary" | ||
| "github.com/golang/snappy" | ||
| "github.com/wader/fq/format" | ||
| "github.com/wader/fq/format/avro/decoders" | ||
| "github.com/wader/fq/format/avro/schema" | ||
| "github.com/wader/fq/format/registry" | ||
| "github.com/wader/fq/pkg/bitio" | ||
| "github.com/wader/fq/pkg/decode" | ||
| "github.com/wader/fq/pkg/scalar" | ||
| "hash/crc32" | ||
| ) | ||
|
|
||
| func init() { | ||
| registry.MustRegister(decode.Format{ | ||
| Name: format.AVRO_OCF, | ||
| Description: "Avro object container file", | ||
| Groups: []string{format.PROBE}, | ||
| DecodeFn: decodeAvroOCF, | ||
| }) | ||
| } | ||
|
|
||
|
|
@@ -62,13 +68,13 @@ func decodeHeader(d *decode.D) HeaderData { | |
| if err != nil { | ||
| d.Fatalf("failed to parse schema: %v", err) | ||
| } | ||
| if codec, ok := meta["avro.codec"]; ok { | ||
| headerData.Codec, ok = codec.(string) | ||
| if !ok { | ||
| d.Fatalf("avro.codec is not a string") | ||
| } | ||
| } else { | ||
| headerData.Codec = "null" | ||
| } | ||
|
|
||
| headerData.Sync, ok = headerRecord["sync"].([]byte) | ||
|
|
@@ -78,7 +84,7 @@ func decodeHeader(d *decode.D) HeaderData { | |
| return headerData | ||
| } | ||
|
|
||
| func decodeAvroOCF(d *decode.D, in interface{}) interface{} { | ||
| header := decodeHeader(d) | ||
|
|
||
| decodeFn, err := decoders.DecodeFnForSchema(header.Schema) | ||
|
|
@@ -92,18 +98,58 @@ func avroDecodeOCF(d *decode.D, in interface{}) interface{} { | |
| return | ||
| } | ||
| size := d.FieldSFn("size", decoders.VarZigZag) | ||
| i := int64(0) | ||
|
|
||
| if header.Codec == "deflate" { | ||
| br := d.FieldRawLen("compressed", size*8) | ||
| bb := &bytes.Buffer{} | ||
| d.MustCopy(bb, flate.NewReader(bitio.NewIOReader(br))) | ||
| d.FieldArrayRootBitBufFn("data", bitio.NewBitReader(bb.Bytes(), -1), func(d *decode.D) { | ||
| for ; i < count; i++ { | ||
| decodeFn("data", d) | ||
| } | ||
| }) | ||
| } else if header.Codec == "snappy" { | ||
| // Everything but last 4 bytes which are the checksum | ||
| n := (size - 4) * 8 | ||
| br := d.FieldRawLen("compressed", n) | ||
| data := make([]byte, size-4) | ||
| if _, err := br.ReadBits(data, n); err != nil { | ||
| d.Fatalf("failed to read snappy compressed data: %v", err) | ||
| } | ||
This comment has been minimized.
Sorry, something went wrong.
This comment has been minimized.
Sorry, something went wrong. 
xentripetal
Author
Contributor
|
||
| decoded, err := snappy.Decode(nil, data) | ||
This comment has been minimized.
Sorry, something went wrong.
This comment has been minimized.
Sorry, something went wrong.
xentripetal
Author
Contributor
|
||
| if err != nil { | ||
| d.Fatalf("failed to decode snappy compressed data: %v", err) | ||
| } | ||
|
|
||
| crc := crc32.ChecksumIEEE(decoded) | ||
| crcB := make([]byte, 4) | ||
| if d.Endian == decode.BigEndian { | ||
| binary.BigEndian.PutUint32(crcB, crc) | ||
| } else { | ||
| binary.LittleEndian.PutUint32(crcB, crc) | ||
| } | ||
| d.FieldRawLen("crc32", 4*8, d.AssertBitBuf(crcB)) | ||
This comment has been minimized.
Sorry, something went wrong.
This comment has been minimized.
Sorry, something went wrong. |
||
| d.FieldArrayRootBitBufFn("data", bitio.NewBitReader(decoded, -1), func(d *decode.D) { | ||
| for ; i < count; i++ { | ||
| decodeFn("data", d) | ||
| } | ||
| }) | ||
| } else if header.Codec == "null" { | ||
| d.FieldArrayLoop("data", func() bool { return i < count }, func(d *decode.D) { | ||
| decodeFn("datum", d) | ||
| i++ | ||
| }) | ||
| } else { | ||
| // Unknown codec, just dump the compressed data. | ||
| d.FieldRawLen("data", size*8, scalar.Description(header.Codec+" encoded")) | ||
| } | ||
| d.FieldRawLen("sync", 16*8, d.AssertBitBuf(header.Sync)) | ||
| }) | ||
|
|
||
| return nil | ||
| } | ||
|
|
||
| func decodeBlock(d *decode.D, decodeFn decoders.DecodeFn) { | ||
|
|
||
| } | ||
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,128 +1,12 @@ | ||
| # Testcase taken from linkedin/goavro https://github.com/linkedin/goavro | ||
| $ fq '.blocks[0]' quickstop-deflate.avro | ||
This comment has been minimized.
Sorry, something went wrong. 
wader
Owner
|
||
| |00 01 02 03 04 05 06 07 08 09 0a 0b 0c 0d 0e 0f|0123456789abcdef|.blocks[0]{}: | ||
| 0x0000|02 0a 44 61 6e 74 65 0a 48 69 63 6b 73 06 28 30|..Dante.Hicks.(0| data[0:602]: | ||
| * |until 0x3ff1.7 (end) (16370) | | | ||
| 0x00120| b4 09 | .. | count: 602 | ||
| 0x00120| be 22 | ." | size: 2207 | ||
| 0x00120| 8d db| ..| compressed: raw bits | ||
| 0x00130|6f 64 ac f9 19 c6 71 f9 37 49 8e 63 1d 55 55 b5|od....q.7I.c.UU.| | ||
| * |until 0x9cc.7 (2207) | | | ||
| 0x009c0| 93 e7 87| ...| sync: raw bits (valid) | ||
| 0x009d0|9e 02 95 d5 9e 4f 58 37 ad b2 a2 ce cd |.....OX7..... | | ||
Binary file not shown.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,13 @@ | ||
| # Generated using https://gist.github.com/xentripetal/c0f1645ee1abd4d25f71896c8d650543 | ||
| $ fq '.blocks[0]' snappy.avro | ||
This comment has been minimized.
Sorry, something went wrong.
wader
Owner
|
||
| |00 01 02 03 04 05 06 07 08 09 0a 0b 0c 0d 0e 0f|0123456789abcdef|.blocks[0]{}: | ||
| 0x000|01 0e 42 00 00 00 00 92 24 49 92 24 49 f2 3f 02|..B.....$I.$I.?.| data[0:10]: | ||
| * |until 0x307.7 (end) (776) | | | ||
| 0x0420|14 |. | count: 10 | ||
| 0x0420| 8c 08 | .. | size: 518 | ||
| 0x0420| 88 06 f0 52 01 0e 42 00 00 00 00 92 24| ...R..B.....$| compressed: raw bits | ||
| 0x0430|49 92 24 49 f2 3f 02 39 04 31 30 00 00 06 02 61|I.$I.?.9.10....a| | ||
| * |until 0x624.7 (514) | | | ||
| 0x0620| 87 b8 fe b6 | .... | crc32: raw bits (valid) | ||
| 0x0620| cc cc 61 31 fd 14 d0| ..a1...| sync: raw bits (valid) | ||
| 0x0630|61 16 b6 0f 9d 30 f4 1b f0| |a....0...| | | ||
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Using
ReadBitsdirectly is usually a bad idea as it works the same way asio.Reader:sReadso it can return short even if there is more data to be read, for example if the underlaying bit reader is abitio.MultiReader(decoding from a "io-list" etc) orbitio.IOBitReadSeeker(reading from aio.Reader).But there are some helpers to hide this for example
bitio.Copy,bitio.ReadFull,d.*CopyBitor you can use abitio.IOReaderin combination withio.Copyetc. It's a bit messy to go between bit and byte land but i want fq to internally use bits as much as possible.BTW the bitio package has README.md explaining a how it works. Actually now i think about that text should probably be moved to be package documentation.