2010YOUY01 commented on code in PR #25371: URL: https://github.com/apache/datafusion/pull/25371#discussion_r4037112087
########## datafusion/physical-plan/src/joins/logical_batch.rs: ########## @@ -0,0 +1,748 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you 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. + +//! [`LogicalBatch`]: a logically contiguous batch stored as a sequence of +//! [`RecordBatch`]es. + +use std::sync::Arc; + +use arrow::array::{Array, ArrayRef, RecordBatch, UInt32Array, new_empty_array}; +use arrow::compute::{TakeOptions, concat, concat_batches, take}; +use arrow::datatypes::SchemaRef; +use datafusion_common::{Result, exec_datafusion_err, exec_err}; + +/// A logically contiguous batch backed by a sequence of [`RecordBatch`]es +/// that share one schema. Methods on this struct accept global row indices. +/// +/// # Example +/// ```text +/// +/// segment 0 (3 rows) segment 1 (2 rows) segment 2 (4 rows) +/// ┌───┬───┬───┐ ┌───┬───┐ ┌───┬───┬───┬───┐ +/// │ a │ b │ c │ │ d │ e │ │ f │ g │ h │ i │ +/// └───┴───┴───┘ └───┴───┘ └───┴───┴───┴───┘ +/// 0 1 2 3 4 5 6 7 8 ◀── global row index +/// ``` +/// +/// # Motivation +/// +/// Joins (e.g. Nested Loop Join) usually buffer all build-side input, and next concatenating +/// them into a contiguous batch, before the next step. It will 2X the memory usage +/// since fragmented batches and final contiguous batch exist at the same time. This +/// struct avoids concatenation step, and helps reduce memory usage by 2X. +/// +/// Avoiding memory concatenating overhead is not the motivation, since it's usually +/// fast and not a bottleneck in real workloads; at the same time single-batch abstraction +/// help simplify join logic. +/// +/// See issue for details: +/// - <https://github.com/apache/datafusion/issues/23076> +/// +/// # TODO +/// It's named 'logical batch' because it's possible to swap the physical layout +/// and keep the same interface for other usages. For example, segments are aligned +/// at the same size, so it achieves O(1) access speed. +#[derive(Debug, Clone)] +pub(crate) struct LogicalBatch { + schema: SchemaRef, + /// The underlying batches, in row order. Empty batches are dropped on + /// construction, so every segment holds at least one row. + segments: Vec<RecordBatch>, + /// `offsets[i]` is the global index of the first row of `segments[i]`; + /// `offsets[segments.len()]` is the total number of rows. + offsets: Vec<usize>, +} + +impl LogicalBatch { + /// Creates a logical batch from `batches`, which must all have `schema`. + /// + /// # Errors + /// + /// Returns an execution error if a batch has a different schema or the + /// total row count overflows. + pub(crate) fn new(schema: SchemaRef, batches: Vec<RecordBatch>) -> Result<Self> { + if batches.iter().any(|batch| batch.schema() != schema) { Review Comment: It's reasonable to allow not 100% equivalent metadata for different batches, the issue is we don't have a project-level spec for how to handle batch-level metadata, and don't know how to test that e2e (this requires use cases that depend on metadata) For instance, what should we do if we want to concat batches with conflicting metadata keys: ``` -- Conflicting metadata keys across batch batch1 metadata: key: config, value: foo batch2 metadata: key: config, value: bar ``` or after left_batch join right_batch, how to propagate metadata to output batch metadata. The existing implementation is likely quite random across the codebase, so I'd prefer to keep it stricter. If someone have application with such metadata, they must find the existing random behavior, and next we can carry out the plan to agree on spec, and figure out how to get it tested, etc. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
