Repository navigation
SNOW-2499292 Add serializable Arrow batches for distributed fetch - #1833
sfc-gh-pfus wants to merge 2 commits into
Conversation
eb4b175 to
8cdcbee
Compare
e006472 to
78d6c6d
Compare
sfc-gh-boler
left a comment
There was a problem hiding this comment.
Just a first pass and some high level commentary. This is a big enough change that it deserves a technical design doc and a PRD.
| // | ||
| // Call ToSerializable before Fetch: after a successful Fetch the in-memory records are | ||
| // released, though the descriptor itself remains valid for re-serialization. | ||
| func (rb *ArrowBatch) ToSerializable() (SerializableArrowBatch, error) { |
There was a problem hiding this comment.
Just "Serialize" or "Marshal" to be consistent with the usual naming scheme.
There was a problem hiding this comment.
But this is a bit different - this is not already serialised, this function just takes not serialisable struct and turns it to the serialisable struct.
| // The returned batch's Fetch downloads (if remote) and transforms the records exactly | ||
| // as a batch obtained directly from GetArrowBatches would. | ||
| // | ||
| // A remote batch (URL set) requires an *http.Client via WithHTTPClient. Inline batches |
There was a problem hiding this comment.
It it requires a param it should be passed directly, not through options.
There was a problem hiding this comment.
It's not required, for inline chunk it is not used at all. But apart from global timeout, http.DefaultClient is good enough. I'm marking it as optional.
| // Expiry: presigned URLs are valid only for a server-controlled window (typically a few | ||
| // hours). A remote SerializableArrowBatch must be shipped and fetched within it. Inline | ||
| // batches (InlineData set) embed their data and never expire. | ||
| type SerializableArrowBatch struct { |
There was a problem hiding this comment.
It'd just wall it "Serialized". ArrowBatch suffix is pointless, because it is present in the package name already.
There was a problem hiding this comment.
But it's not serialised, it's only able to be serialised. I can shorten the name, it may be more go-like, but is arrowbatches.Serializable fine?
| RowTypes []query.ExecResponseRowType `json:"rowTypes"` | ||
| // TimezoneName is the batch's timezone name (loc.String()). | ||
| TimezoneName string `json:"timezoneName,omitempty"` | ||
| // TimezoneOffsetSeconds is the batch's timezone offset, used as a fallback when the |
There was a problem hiding this comment.
What's the point of TZName if we give the offset already?
There was a problem hiding this comment.
Because Go is not Snowflake :) Snowflake uses offset only for timestamps, but to create a correct timezone in Go you need both. It is needed for LTZ.
| TimezoneName string `json:"timezoneName,omitempty"` | ||
| // TimezoneOffsetSeconds is the batch's timezone offset, used as a fallback when the | ||
| // worker cannot resolve TimezoneName (e.g. fixed-offset zones, or missing tzdata). | ||
| TimezoneOffsetSeconds int `json:"timezoneOffsetSeconds,omitempty"` |
There was a problem hiding this comment.
TZs don't have offsets in seconds. That's confusing. Minutes would be fine.
There was a problem hiding this comment.
But Go API requires seconds to create a location:
// FixedZone returns a [Location] that always uses
// the given zone name and offset (seconds east of UTC).
func FixedZone(name string, offset int) *Location {
| } | ||
| resp, err := client.Do(req) | ||
| if err != nil { | ||
| return nil, fmt.Errorf("arrow: fetching chunk: %w", err) |
There was a problem hiding this comment.
is this going to print the URL containing keys (presigned)?
There was a problem hiding this comment.
It shouldn't, it goes through our secret detector.
| for k, v := range headers { | ||
| req.Header.Set(k, v) | ||
| } | ||
| resp, err := client.Do(req) |
There was a problem hiding this comment.
what's the retry and cancelation story?
There was a problem hiding this comment.
Re: cancellation - this request takes context, so it should honour cancellations as usual.
Re: retry - we have an ugly retry mechanism, but it is in our main package, not available here. I can move it to internal package (http?), but I'd prefer doing this in another PR.
| if inlineBytes, err := base64.StdEncoding.DecodeString(scd.RowSet.RowSetBase64); err == nil { | ||
| batch.Descriptor = ia.ChunkDescriptor{InlineData: inlineBytes} | ||
| } else { | ||
| logger.WithContext(scd.ctx).Warnf("failed to decode RowSetBase64 for arrow batch serialization: %v", err) |
There was a problem hiding this comment.
So we log an error, but there is no signalling to the caller it seems.
There was a problem hiding this comment.
Good catch. I also think that there's no point in decoding it here, since this field is used only for serialising it later. So I kept base64 instead of byte array.
| // field is kept for forward compatibility with the streaming/JSON result path. | ||
| const arrowFormatName = "arrow" | ||
|
|
||
| // SerializableArrowBatch is a portable, self-contained description of an ArrowBatch. |
There was a problem hiding this comment.
Do we even want to expose those fields? They seem internal to Marsha/Unmarshal.
There was a problem hiding this comment.
Yes, this is needed for JSON serialisation.
| // hours). A remote SerializableArrowBatch must be shipped and fetched within it. Inline | ||
| // batches (InlineData set) embed their data and never expire. | ||
| type SerializableArrowBatch struct { | ||
| // Format is the chunk payload format; currently always "arrow". |
There was a problem hiding this comment.
What does this mean specifically? In other words, when is this going to change?
If we want to provide a contract between orchestrators and executors (distributed system), we need a version number here.
What about orchestration between different drivers? Is the plan to support it?
There was a problem hiding this comment.
This means, that this should be always Arrow, not JSON.
This is never an external executors, it should be always handled by a drivers. So all these should be an implementation detail.
About other drivers - only Go driver supports arrow batches in this form. We can think about improvements in UD, if needed.
78d6c6d to
da861b8
Compare
da861b8 to
a9ad80d
Compare
| desc := ia.ChunkDescriptor{ | ||
| URL: s.URL, | ||
| Headers: s.Headers, | ||
| InlineDataBase64: s.InlineData, | ||
| UncompressedSize: s.UncompressedSize, | ||
| } |
There was a problem hiding this comment.
🟠 MEDIUM · OpenChunkStream follows arbitrary attacker-influenced URL from a deserialized descriptor with no scheme/host allowli · CWE-918

TL;DR
Adds
SerializableArrowBatchwithArrowBatch.ToSerializableandSerializableArrowBatch.ToArrowBatchto enable distributed fetch of Arrow result batches across machines with no Snowflake connection.What changed?
arrowbatches.SerializableArrowBatch, a portable JSON-serializable descriptor that captures everything needed to fetch a result chunk: the presigned URL, SSE-C headers, inline IPC bytes (for the first batch), row count, column metadata, timezone, and conversion options.ArrowBatch.ToSerializable()to convert a live batch handle into aSerializableArrowBatchbefore fetching.SerializableArrowBatch.ToArrowBatch(opts ...Option)to reconstruct a fully functionalArrowBatchon a worker node using only an injected*http.Client— no Snowflake session required.chunkDownloadHeadersinchunk_downloader.goas a shared helper so the same SSE-C/GCS/Azure header logic is used by both the live download path and the serializable descriptor construction.BatchRaw.Descriptor(a newChunkDescriptorfield onia.BatchRaw) inrows.gofor both remote chunks (URL + headers) and the inline first batch (base64-decodedRowSetBase64).internal/arrow/download.gowithOpenChunkStream,DecodeIPCRecords, andDownloadBatchRecords— low-level primitives that perform HTTP download and IPC decoding without any driver state, used by the reconstructed batch'sDownloadclosure.ConversionOptionsfields so they survive serialization as part ofSerializableArrowBatch.cmd/arrow/serializablebatchesexample demonstrating the full producer → serialize → worker → fetch flow.doc.gowith documentation covering the distributed fetch pattern, security considerations (presigned URLs and SSE-C keys are sensitive), and expiry caveats.How to test?
Unit tests in
arrowbatches/serializable_test.gocover:RowTypesand allConversionOptionsflags.loadLocationbehavior for fixed-offset,Local, named IANA, and empty timezone names.ConversionOptions(e.g.HigherPrecision) carried in the serialized form are actually applied during fetch, not just round-tripped as data.Integration tests in
arrowbatches/serialization_integration_test.go(require a live account) cover:Why make this change?
Arrow batches returned by
GetArrowBatchesare live handles tied to the Snowflake connection that produced them. Distributed compute frameworks (Spark, Hadoop, etc.) need to partition a large result on a coordinator node and download chunks in parallel on worker nodes that have no Snowflake session.SerializableArrowBatchprovides the seam for this: the coordinator serializes batch descriptors (which embed presigned URLs and decryption keys), ships them to workers, and each worker fetches its chunk independently using only a standard HTTP client.