dongjoon-hyun commented on code in PR #57899:
URL: https://github.com/apache/spark/pull/57899#discussion_r3754392229
##########
python/pyspark/worker.py:
##########
@@ -3122,6 +3222,257 @@ def func(split_index: int, data:
Iterator[pa.RecordBatch]) -> Iterator[pa.Record
# profiling is not supported for UDF
return func, None, ser, ser
+ if eval_type in (
+ PythonEvalType.SQL_SCALAR_PANDAS_ELEMENTWISE_UDF,
+ PythonEvalType.SQL_SCALAR_ARROW_ELEMENTWISE_UDF,
+ ):
+ from pyspark.sql.pandas.utils import require_minimum_pyarrow_version
+
+ require_minimum_pyarrow_version()
+
+ import pyarrow as pa
+ import pyarrow.compute as pc
+
+ # A scalar pandas or Arrow UDF lifted out of a higher-order function's
lambda by
+ # ExtractPythonUDFFromLambda. Each argument arrives as ``array<T>``
aligned with the
+ # iterated array. We flatten each list column to its element column,
run the *vectorized*
+ # function once over that flat column (so it still receives a pandas
Series / DataFrame or a
+ # pa.Array, its native contract), then re-nest the flat result to
``array<R>`` using the
+ # input's offsets - one row in, one row out, one Python round trip per
batch.
+ is_pandas = eval_type ==
PythonEvalType.SQL_SCALAR_PANDAS_ELEMENTWISE_UDF
+
+ udf_infos = []
+ for udf_func, udf_args_offsets, udf_kwargs_offsets, udf_return_type in
udfs:
+ wrapped_func, args_kwargs_offsets = wrap_kwargs_support(
+ udf_func, udf_args_offsets, udf_kwargs_offsets
+ )
+ # The UDF returns one value per element, so its declared return
type is the element
+ # type of the ``array<R>`` this operator produces.
+ arrow_element_type = to_arrow_type(
+ udf_return_type, timezone="UTC",
prefers_large_types=runner_conf.use_large_var_types
+ )
+ udf_infos.append(
+ (wrapped_func, args_kwargs_offsets, udf_return_type,
arrow_element_type)
+ )
+ col_names = [f"_{i}" for i in range(len(udfs))]
+
+ if is_pandas:
+ import pandas as pd
+
+ # Each argument is ``array<T>``; the vectorized function must see the
element type ``T``.
+ element_types = [f.dataType.elementType for f in eval_conf.input_type]
+
+ def func(split_index: int, data: Iterator[pa.RecordBatch]) ->
Iterator[pa.RecordBatch]:
+ for input_batch in data:
+ # Flatten each list column to its element array once per batch
and share it across
+ # fused UDFs (the 102 path does the same), rather than
re-flattening per UDF.
+ flat_columns = [
+ _elementwise_flatten_column(
+ col.flatten(), element_types[i], is_pandas, runner_conf
+ )
+ for i, col in enumerate(input_batch.itercolumns())
+ ]
+
+ output_arrays = []
+ for wrapped_func, offsets, return_type, arrow_element_type in
udf_infos:
+ # Re-nest by this UDF's first argument's shape. Different
UDFs in one operator
+ # may iterate differently shaped arrays, so each re-nests
by its own argument.
+ shape = input_batch.column(offsets[0])
+ shape_lengths = pc.list_value_length(shape).to_pylist()
+ total_elements = sum(n for n in shape_lengths if n is not
None)
+
+ result = wrapped_func(*[flat_columns[o] for o in offsets])
+ if is_pandas:
+ if not hasattr(result, "__len__"):
+ pd_type = (
+ "pandas.DataFrame"
+ if isinstance(return_type, StructType)
+ else "pandas.Series"
+ )
+ raise PySparkTypeError(
+ errorClass="UDF_RETURN_TYPE",
+ messageParameters={
+ "expected": pd_type,
+ "actual": type(result).__name__,
+ },
+ )
+ # struct return type must be a DataFrame (matches the
base pandas path).
+ if isinstance(return_type, StructType) and not
isinstance(
+ result, pd.DataFrame
+ ):
+ raise PySparkValueError(
+ "Invalid return type. Please make sure that
the UDF returns a "
+ "pandas.DataFrame when the specified return
type is StructType."
+ )
+ # Verify the flat length before re-nesting so a
wrong-length result raises the
+ # friendly RESULT_ROWS_MISMATCH rather than an opaque
pyarrow offset error.
+ verify_result_row_count(len(result), total_elements)
+
+ flat_arr = _elementwise_result_to_arrow(
+ result, return_type, arrow_element_type, is_pandas,
runner_conf
+ )
+ nested = _elementwise_renest(
+ flat_arr, shape_lengths,
pa.types.is_large_list(shape.type)
+ )
+ output_arrays.append(nested)
+
+ yield pa.RecordBatch.from_arrays(output_arrays, col_names)
+
+ # profiling is not supported for UDF
+ return func, None, ser, ser
+
+ if eval_type in (
+ PythonEvalType.SQL_SCALAR_PANDAS_ITER_ELEMENTWISE_UDF,
+ PythonEvalType.SQL_SCALAR_ARROW_ITER_ELEMENTWISE_UDF,
+ ):
+ from pyspark.sql.pandas.utils import require_minimum_pyarrow_version
+
+ require_minimum_pyarrow_version()
+
+ import collections
+
+ import pyarrow as pa
+ import pyarrow.compute as pc
+
+ is_pandas = eval_type ==
PythonEvalType.SQL_SCALAR_PANDAS_ITER_ELEMENTWISE_UDF
+ if is_pandas:
+ import pandas as pd
+
+ assert num_udfs == 1, "One SCALAR_*_ITER_ELEMENTWISE UDF expected
here."
+ udf_func, args_offsets, kwargs_offsets, return_type = udfs[0]
+ assert not kwargs_offsets, "Iterator UDFs do not take keyword
arguments."
+
+ # A scalar iterator UDF (pandas or Arrow) lifted out of a higher-order
function's lambda.
+ # The user function keeps its iterator contract: it consumes an
iterator of batches and
+ # yields an iterator of batches, one output value per input value. We
preserve that by
+ # feeding it the *flattened* elements of each input batch and, since
the JVM joins UDF
+ # output to input positionally by row (one ``array<R>`` per input
``array<T>`` row, in
+ # order), buffering a FIFO of the per-row element counts to re-group
the streamed flat
+ # results back into arrays. Output batch boundaries need not match
input ones.
+ # Each argument is ``array<T>``; the pandas function must see each
argument's own element
+ # type ``T`` (arguments may differ, e.g. an outer column repeated into
an aligned array).
+ element_types = [f.dataType.elementType for f in eval_conf.input_type]
+ arrow_element_type = to_arrow_type(
+ return_type, timezone="UTC",
prefers_large_types=runner_conf.use_large_var_types
+ )
+ is_large = None # set from the first input batch; the list width is
uniform per column.
+
+ def func(split_index: int, data: Iterator[pa.RecordBatch]) ->
Iterator[pa.RecordBatch]:
+ # FIFO of per-row element counts (None for a null array) awaiting
their flat results,
+ # and the flat elements produced so far but not yet enough to
complete the head shapes.
+ pending_shapes: "collections.deque" = collections.deque()
+ num_input_elements = 0
+
+ def extract_flat(batch: pa.RecordBatch):
+ nonlocal is_large, num_input_elements
+ shape = batch.column(args_offsets[0])
+ if is_large is None:
+ is_large = pa.types.is_large_list(shape.type)
+ pending_shapes.append(pc.list_value_length(shape).to_pylist())
+ # Flatten each argument to its element column; the user
function sees the flat
+ # elements as a pandas Series / DataFrame (pandas) or a
pa.Array (Arrow). Each
+ # argument is converted with its own element type.
+ flat_cols = [
+ _elementwise_flatten_column(
+ batch.column(o).flatten(), element_types[o],
is_pandas, runner_conf
+ )
+ for o in args_offsets
+ ]
+ num_input_elements += len(flat_cols[0])
+ return flat_cols[0] if len(flat_cols) == 1 else
tuple(flat_cols)
+
+ flat_args_iter = map(extract_flat, data)
+
+ if not is_pandas:
+ verified_iter = verify_return_type(
+ udf_func(flat_args_iter),
+ Iterator[pa.Array], # type: ignore[type-abstract]
+ )
+ else:
+ pandas_iter_type = (
+ Iterator[pd.DataFrame]
+ if isinstance(return_type, StructType)
+ else Iterator[pd.Series]
+ )
+ verified_iter = verify_return_type(udf_func(flat_args_iter),
pandas_iter_type)
+
+ # Buffer the streamed flat element results and emit an
``array<R>`` row as soon as the
+ # shape at the head of the FIFO is fully covered. A row whose
length is 0 (an empty
+ # array) or None (a null array) needs no elements, so it is
emitted immediately even
+ # before any chunk arrives - this matters when a whole partition
is empty/null arrays
+ # and the UDF yields nothing, otherwise those rows would be
dropped by the positional
+ # JVM join. Chunks are held in a list and concatenated only when a
shape spans more than
+ # one, so a UDF that yields once per input batch (the common case)
never re-copies the
+ # buffer. ``empty_type`` supplies the element type for a
zero-length emit: the first
+ # chunk's type once seen (the pandas flavor types timestamps with
the session timezone),
+ # else the UTC-typed ``arrow_element_type``. A partition that only
ever emits
+ # zero-length rows never mixes the two, so the stream schema stays
consistent.
+ pending_chunks: "list" = []
+ pending_len = 0
+ empty_type = arrow_element_type
+ num_output_elements = 0
+
+ def emit_ready():
+ nonlocal pending_chunks, pending_len
+ while pending_shapes:
+ lengths = pending_shapes[0]
+ needed = sum(n for n in lengths if n is not None)
+ if needed > pending_len:
+ break
+ pending_shapes.popleft()
+ if needed == 0:
+ flat = pa.nulls(0, type=empty_type)
+ else:
+ combined = (
+ pending_chunks[0]
+ if len(pending_chunks) == 1
+ else pa.concat_arrays(pending_chunks)
+ )
+ flat = combined.slice(0, needed)
+ remainder = combined.slice(needed)
+ pending_chunks = [remainder] if len(remainder) else []
+ pending_len -= needed
+ nested = _elementwise_renest(flat, lengths, bool(is_large))
+ yield pa.RecordBatch.from_arrays([nested], ["_0"])
+
+ def process_results():
+ nonlocal pending_chunks, pending_len, empty_type,
num_output_elements
+ for result in verified_iter:
+ if is_pandas:
+ verify_pandas_result(
+ result,
+ return_type,
+ assign_cols_by_name=True,
+ truncate_return_schema=True,
+ )
+ chunk = _elementwise_result_to_arrow(
+ result, return_type, arrow_element_type, is_pandas,
runner_conf
+ )
+ num_output_elements += len(chunk)
+ # Fail fast if the UDF over-produces, before the buffer
grows unbounded (the
+ # base iterator paths do the same via
verify_output_row_limit).
+ if num_output_elements > num_input_elements:
+ raise PySparkRuntimeError(
+ errorClass="OUTPUT_EXCEEDS_INPUT_ROWS",
messageParameters={}
+ )
+ if len(chunk):
+ pending_chunks.append(chunk)
+ pending_len += len(chunk)
+ empty_type = chunk.type
Review Comment:
`empty_type` is only updated for non-empty chunks, so a zero-length chunk
yielded for an input batch holding only empty/null arrays leaves it at the
UTC-typed `arrow_element_type`. The rows for that batch are emitted immediately
with that UTC type, fixing the output stream's schema; a later real chunk from
the pandas flavor is typed with the *session* timezone, so writing the next
output batch fails in `ArrowStreamSerializer.dump_stream` with
`pyarrow.lib.ArrowInvalid: Tried to write record batch with different schema`.
Reproduced with a timestamp-returning pandas iterator UDF, session timezone
`America/Los_Angeles`, `spark.sql.execution.arrow.maxRecordsPerBatch=1`, and a
single partition `[([],), ([1],)]` (see the suggested test in
`test_udf_in_higher_order_function.py`). A zero-length chunk still carries the
flavor's element type, so taking the type from every chunk keeps the stream
schema consistent. Verified the repro passes with this change and the existing
iterator tests (`test_scalar_iter_udf_over_all_empty_and_null_partition`,
`test_scalar_pandas_iter_udf_timestamp_return_type`, and the pandas/Arrow
iterator basics) still pass.
```suggestion
# Even a zero-length chunk carries the flavor's element
type (the pandas
# flavor types timestamps with the session timezone), so
always take it:
# otherwise rows emitted for an all-empty batch before
the first non-empty
# chunk would use the UTC-typed default and disagree
with later batches,
# breaking the output stream's single-schema contract.
empty_type = chunk.type
if len(chunk):
pending_chunks.append(chunk)
pending_len += len(chunk)
```
##########
python/pyspark/worker.py:
##########
@@ -3122,6 +3222,257 @@ def func(split_index: int, data:
Iterator[pa.RecordBatch]) -> Iterator[pa.Record
# profiling is not supported for UDF
return func, None, ser, ser
+ if eval_type in (
+ PythonEvalType.SQL_SCALAR_PANDAS_ELEMENTWISE_UDF,
+ PythonEvalType.SQL_SCALAR_ARROW_ELEMENTWISE_UDF,
+ ):
+ from pyspark.sql.pandas.utils import require_minimum_pyarrow_version
+
+ require_minimum_pyarrow_version()
+
+ import pyarrow as pa
+ import pyarrow.compute as pc
+
+ # A scalar pandas or Arrow UDF lifted out of a higher-order function's
lambda by
+ # ExtractPythonUDFFromLambda. Each argument arrives as ``array<T>``
aligned with the
+ # iterated array. We flatten each list column to its element column,
run the *vectorized*
+ # function once over that flat column (so it still receives a pandas
Series / DataFrame or a
+ # pa.Array, its native contract), then re-nest the flat result to
``array<R>`` using the
+ # input's offsets - one row in, one row out, one Python round trip per
batch.
+ is_pandas = eval_type ==
PythonEvalType.SQL_SCALAR_PANDAS_ELEMENTWISE_UDF
+
+ udf_infos = []
+ for udf_func, udf_args_offsets, udf_kwargs_offsets, udf_return_type in
udfs:
+ wrapped_func, args_kwargs_offsets = wrap_kwargs_support(
+ udf_func, udf_args_offsets, udf_kwargs_offsets
+ )
+ # The UDF returns one value per element, so its declared return
type is the element
+ # type of the ``array<R>`` this operator produces.
+ arrow_element_type = to_arrow_type(
+ udf_return_type, timezone="UTC",
prefers_large_types=runner_conf.use_large_var_types
+ )
+ udf_infos.append(
+ (wrapped_func, args_kwargs_offsets, udf_return_type,
arrow_element_type)
+ )
+ col_names = [f"_{i}" for i in range(len(udfs))]
+
+ if is_pandas:
+ import pandas as pd
+
+ # Each argument is ``array<T>``; the vectorized function must see the
element type ``T``.
+ element_types = [f.dataType.elementType for f in eval_conf.input_type]
+
+ def func(split_index: int, data: Iterator[pa.RecordBatch]) ->
Iterator[pa.RecordBatch]:
+ for input_batch in data:
+ # Flatten each list column to its element array once per batch
and share it across
+ # fused UDFs (the 102 path does the same), rather than
re-flattening per UDF.
+ flat_columns = [
+ _elementwise_flatten_column(
+ col.flatten(), element_types[i], is_pandas, runner_conf
+ )
+ for i, col in enumerate(input_batch.itercolumns())
+ ]
+
+ output_arrays = []
+ for wrapped_func, offsets, return_type, arrow_element_type in
udf_infos:
+ # Re-nest by this UDF's first argument's shape. Different
UDFs in one operator
+ # may iterate differently shaped arrays, so each re-nests
by its own argument.
+ shape = input_batch.column(offsets[0])
+ shape_lengths = pc.list_value_length(shape).to_pylist()
+ total_elements = sum(n for n in shape_lengths if n is not
None)
+
+ result = wrapped_func(*[flat_columns[o] for o in offsets])
+ if is_pandas:
+ if not hasattr(result, "__len__"):
+ pd_type = (
+ "pandas.DataFrame"
+ if isinstance(return_type, StructType)
+ else "pandas.Series"
+ )
+ raise PySparkTypeError(
+ errorClass="UDF_RETURN_TYPE",
+ messageParameters={
+ "expected": pd_type,
+ "actual": type(result).__name__,
+ },
+ )
+ # struct return type must be a DataFrame (matches the
base pandas path).
+ if isinstance(return_type, StructType) and not
isinstance(
+ result, pd.DataFrame
+ ):
+ raise PySparkValueError(
+ "Invalid return type. Please make sure that
the UDF returns a "
+ "pandas.DataFrame when the specified return
type is StructType."
+ )
+ # Verify the flat length before re-nesting so a
wrong-length result raises the
+ # friendly RESULT_ROWS_MISMATCH rather than an opaque
pyarrow offset error.
+ verify_result_row_count(len(result), total_elements)
+
+ flat_arr = _elementwise_result_to_arrow(
+ result, return_type, arrow_element_type, is_pandas,
runner_conf
+ )
+ nested = _elementwise_renest(
+ flat_arr, shape_lengths,
pa.types.is_large_list(shape.type)
+ )
+ output_arrays.append(nested)
+
+ yield pa.RecordBatch.from_arrays(output_arrays, col_names)
+
+ # profiling is not supported for UDF
+ return func, None, ser, ser
+
+ if eval_type in (
+ PythonEvalType.SQL_SCALAR_PANDAS_ITER_ELEMENTWISE_UDF,
+ PythonEvalType.SQL_SCALAR_ARROW_ITER_ELEMENTWISE_UDF,
+ ):
+ from pyspark.sql.pandas.utils import require_minimum_pyarrow_version
+
+ require_minimum_pyarrow_version()
+
+ import collections
+
+ import pyarrow as pa
+ import pyarrow.compute as pc
+
+ is_pandas = eval_type ==
PythonEvalType.SQL_SCALAR_PANDAS_ITER_ELEMENTWISE_UDF
+ if is_pandas:
+ import pandas as pd
+
+ assert num_udfs == 1, "One SCALAR_*_ITER_ELEMENTWISE UDF expected
here."
+ udf_func, args_offsets, kwargs_offsets, return_type = udfs[0]
+ assert not kwargs_offsets, "Iterator UDFs do not take keyword
arguments."
+
+ # A scalar iterator UDF (pandas or Arrow) lifted out of a higher-order
function's lambda.
+ # The user function keeps its iterator contract: it consumes an
iterator of batches and
+ # yields an iterator of batches, one output value per input value. We
preserve that by
+ # feeding it the *flattened* elements of each input batch and, since
the JVM joins UDF
+ # output to input positionally by row (one ``array<R>`` per input
``array<T>`` row, in
+ # order), buffering a FIFO of the per-row element counts to re-group
the streamed flat
+ # results back into arrays. Output batch boundaries need not match
input ones.
+ # Each argument is ``array<T>``; the pandas function must see each
argument's own element
+ # type ``T`` (arguments may differ, e.g. an outer column repeated into
an aligned array).
+ element_types = [f.dataType.elementType for f in eval_conf.input_type]
+ arrow_element_type = to_arrow_type(
+ return_type, timezone="UTC",
prefers_large_types=runner_conf.use_large_var_types
+ )
+ is_large = None # set from the first input batch; the list width is
uniform per column.
+
+ def func(split_index: int, data: Iterator[pa.RecordBatch]) ->
Iterator[pa.RecordBatch]:
+ # FIFO of per-row element counts (None for a null array) awaiting
their flat results,
+ # and the flat elements produced so far but not yet enough to
complete the head shapes.
+ pending_shapes: "collections.deque" = collections.deque()
+ num_input_elements = 0
+
+ def extract_flat(batch: pa.RecordBatch):
+ nonlocal is_large, num_input_elements
+ shape = batch.column(args_offsets[0])
+ if is_large is None:
+ is_large = pa.types.is_large_list(shape.type)
+ pending_shapes.append(pc.list_value_length(shape).to_pylist())
+ # Flatten each argument to its element column; the user
function sees the flat
+ # elements as a pandas Series / DataFrame (pandas) or a
pa.Array (Arrow). Each
+ # argument is converted with its own element type.
+ flat_cols = [
+ _elementwise_flatten_column(
+ batch.column(o).flatten(), element_types[o],
is_pandas, runner_conf
+ )
+ for o in args_offsets
+ ]
+ num_input_elements += len(flat_cols[0])
+ return flat_cols[0] if len(flat_cols) == 1 else
tuple(flat_cols)
+
+ flat_args_iter = map(extract_flat, data)
+
+ if not is_pandas:
+ verified_iter = verify_return_type(
+ udf_func(flat_args_iter),
+ Iterator[pa.Array], # type: ignore[type-abstract]
+ )
+ else:
+ pandas_iter_type = (
+ Iterator[pd.DataFrame]
+ if isinstance(return_type, StructType)
+ else Iterator[pd.Series]
+ )
+ verified_iter = verify_return_type(udf_func(flat_args_iter),
pandas_iter_type)
+
+ # Buffer the streamed flat element results and emit an
``array<R>`` row as soon as the
+ # shape at the head of the FIFO is fully covered. A row whose
length is 0 (an empty
+ # array) or None (a null array) needs no elements, so it is
emitted immediately even
+ # before any chunk arrives - this matters when a whole partition
is empty/null arrays
+ # and the UDF yields nothing, otherwise those rows would be
dropped by the positional
+ # JVM join. Chunks are held in a list and concatenated only when a
shape spans more than
+ # one, so a UDF that yields once per input batch (the common case)
never re-copies the
+ # buffer. ``empty_type`` supplies the element type for a
zero-length emit: the first
+ # chunk's type once seen (the pandas flavor types timestamps with
the session timezone),
+ # else the UTC-typed ``arrow_element_type``. A partition that only
ever emits
+ # zero-length rows never mixes the two, so the stream schema stays
consistent.
+ pending_chunks: "list" = []
+ pending_len = 0
+ empty_type = arrow_element_type
+ num_output_elements = 0
+
+ def emit_ready():
+ nonlocal pending_chunks, pending_len
+ while pending_shapes:
+ lengths = pending_shapes[0]
+ needed = sum(n for n in lengths if n is not None)
+ if needed > pending_len:
+ break
+ pending_shapes.popleft()
+ if needed == 0:
+ flat = pa.nulls(0, type=empty_type)
+ else:
+ combined = (
+ pending_chunks[0]
+ if len(pending_chunks) == 1
+ else pa.concat_arrays(pending_chunks)
+ )
+ flat = combined.slice(0, needed)
+ remainder = combined.slice(needed)
+ pending_chunks = [remainder] if len(remainder) else []
+ pending_len -= needed
+ nested = _elementwise_renest(flat, lengths, bool(is_large))
+ yield pa.RecordBatch.from_arrays([nested], ["_0"])
+
+ def process_results():
+ nonlocal pending_chunks, pending_len, empty_type,
num_output_elements
+ for result in verified_iter:
+ if is_pandas:
+ verify_pandas_result(
+ result,
+ return_type,
+ assign_cols_by_name=True,
+ truncate_return_schema=True,
+ )
+ chunk = _elementwise_result_to_arrow(
+ result, return_type, arrow_element_type, is_pandas,
runner_conf
+ )
+ num_output_elements += len(chunk)
+ # Fail fast if the UDF over-produces, before the buffer
grows unbounded (the
+ # base iterator paths do the same via
verify_output_row_limit).
+ if num_output_elements > num_input_elements:
+ raise PySparkRuntimeError(
+ errorClass="OUTPUT_EXCEEDS_INPUT_ROWS",
messageParameters={}
Review Comment:
`empty_type` is only updated for non-empty chunks, so a zero-length chunk
yielded for an input batch holding only empty/null arrays leaves it at the
UTC-typed `arrow_element_type`. The rows for that batch are emitted immediately
with that UTC type, fixing the output stream's schema; a later real chunk from
the pandas flavor is typed with the *session* timezone, so writing the next
output batch fails in `ArrowStreamSerializer.dump_stream` with
`pyarrow.lib.ArrowInvalid: Tried to write record batch with different schema`.
Reproduced with a timestamp-returning pandas iterator UDF, session timezone
`America/Los_Angeles`, `spark.sql.execution.arrow.maxRecordsPerBatch=1`, and a
single partition `[([],), ([1],)]` (see the suggested test in
`test_udf_in_higher_order_function.py`). A zero-length chunk still carries the
flavor's element type, so taking the type from every chunk keeps the stream
schema consistent. Verified the repro passes with this change and the existing
iterator tests (`test_scalar_iter_udf_over_all_empty_and_null_partition`,
`test_scalar_pandas_iter_udf_timestamp_return_type`, and the pandas/Arrow
iterator basics) still pass.
```suggestion
# Even a zero-length chunk carries the flavor's element
type (the pandas
# flavor types timestamps with the session timezone), so
always take it:
# otherwise rows emitted for an all-empty batch before
the first non-empty
# chunk would use the UTC-typed default and disagree
with later batches,
# breaking the output stream's single-schema contract.
empty_type = chunk.type
if len(chunk):
pending_chunks.append(chunk)
pending_len += len(chunk)
```
--
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]
