[analytics-engine] wire PPL TAKE / FIRST / LAST / LIST / VALUES end-to-end (#21731)

* analytics-framework: extend AggregateFunction SPI for state-expanding aggregates

Adds five enum entries for PPL TAKE / FIRST / LAST / LIST / VALUES and an
input-parameterised IntermediateField type resolver. Existing fixed-shape
aggregates (HLL Binary sketch, COUNT Int64 counter) keep their constant
intermediate types via IntermediateTypeResolver.fixed(); state-expanding
aggregates whose FINAL state shape derives from arg 0 use
IntermediateTypeResolver.passThroughArg0() so the planner can resolve the
exchange column type from the actual call's arg type rather than a constant.

Internalises the Arrow → Calcite type mapping (previously in the planner
module's ArrowCalciteTypes) so the SPI is self-contained.

Signed-off-by: Sandesh Kumar <sandeshkr419@gmail.com>

* analytics-backend-datafusion: wire 5 PPL state-expanding aggregates end-to-end

Wires PPL TAKE / FIRST / LAST / LIST / VALUES through the DataFusion backend:

* Rust UDAF for take(field, n) at rust/src/udaf/take.rs. Limit is read from a
  Substrait Literal when present, or from row 0 of arg 1 when Calcite has
  materialised n as a constant Project column. Default n = 10. State is a
  bounded List<elem> emitted from state(); coordinator-side merge_batch
  concatenates and re-truncates. UDAF registered on every SessionContext
  via crate::udaf::register_all.

* Substrait extension entries in opensearch_aggregate_functions.yaml for
  take, first_value, last_value, and array_agg. take's 2-arg shape uses
  distinct any1/any2 generics so isthmus's wildcard binding accepts
  mismatched arg types (state list<elem> + integer n).

* DataFusionFragmentConvertor: LOCAL_TAKE_OP / LOCAL_FIRST_OP / LOCAL_LAST_OP /
  LOCAL_ARRAY_AGG_OP stub SqlAggFunctions plus a pre-emit RelShuttle that
  rewrites PPL aggregation calls (TAKE / FIRST / LAST / LIST / VALUES) onto
  the stubs so isthmus's AggregateFunctionConverter resolves them through
  ADDITIONAL_AGGREGATE_SIGS to the YAML extension names. LIST/VALUES rebuild
  the call's return type as ARRAY<actual-arg0> to override PPL's lossy
  STRING_ARRAY which would otherwise force ARRAY<VARCHAR>.

* SubstraitPlanRewriter.visit(Aggregate): post-emit pass that inlines
  Project-column literals into Aggregate.Measure args. Required because
  Calcite's RelBuilder.aggregate auto-projects literal aggregate-args as
  $f1 columns, leaving the substrait emit with FieldRefs instead of
  Literals.

Signed-off-by: Sandesh Kumar <sandeshkr419@gmail.com>

* analytics-engine: distributed planner rewrites for PPL state-expanding aggregates

Three planner-level changes that compose to make TAKE / FIRST / LAST / LIST /
VALUES correct across shards:

1. Move AggregateCallAnnotation out of AggregateCall.rexList onto an
   OpenSearchAggregate side-map keyed by call index. Calcite's
   AggCallBinding.preOperands treats every rexList entry as a leading
   operand for inferReturnType, which corrupts return-type inference for
   PPL operators that read getOperandType(0) (ARG0_ARRAY on TAKE / LIST /
   VALUES would double-wrap). Storing annotations out-of-band on
   OpenSearchAggregate sidesteps the issue without touching PPL. Drops
   the AGG_CALL_ANNOTATION marker class and the annotation-stripping
   logic in stripAnnotations / copyResolved.

2. DistributedAggregateRewriter: drive intermediate-field types through
   the SPI's IntermediateTypeResolver (replacing the static ArrowCalciteTypes
   helper, which is removed). For STATE_EXPANDING + APPROXIMATE engine-native-merge
   aggregates, pin the FINAL aggCall's explicit return type to the StageInputScan
   column type so substrait declares what Rust's derive_schema_from_partial_plan
   produces — PPL's stock ReturnTypes are wrong here (ARG0_ARRAY double-wraps,
   STRING_ARRAY ignores the actual element type).

3. Cross-shard literal-arg forwarding for TAKE's N. OpenSearchAggregateSplitRule
   captures any RexLiteral aggregate-args from the original SINGLE aggregate's
   underlying Project and stashes them on the FINAL OpenSearchAggregate as a
   side-map. DistributedAggregateRewriter (Phase 2b) wraps the FINAL's
   StageInputScan in an OpenSearchProject that re-creates each captured literal
   as a constant column, and rebuilds the FINAL aggCall's argList to
   [stateColIdx, ...litColIdxs]. The convertor's existing SubstraitPlanRewriter
   inliner then emits the literals as Substrait Literal expressions. The
   producer-side PARTIAL stage still consumes N inside its own accumulator;
   without this, FINAL would re-aggregate without N and fall back to the
   default limit.

IT cluster runs with -da:org.apache.calcite... so Calcite's typeMatchesInferred
assertion is silenced — post #21690 the wire schema is derived in Rust, not
Java, and PPL operators with non-idempotent return-type inference would
otherwise trip this assertion on the FINAL side. Production runs without -ea.

CoordinatorReduceIT: ten new tests (single-shard + cross-shard for each of
the five aggregates). LIST/VALUES across-shards remain @AwaitsFix until the
PPL frontend stops declaring STRING_ARRAY for them.

Signed-off-by: Sandesh Kumar <sandeshkr419@gmail.com>

* analytics-engine: cross-shard LIST and VALUES via list_merge UDAF

Two stacked issues blocked cross-shard LIST/VALUES:

1. PPL declares LIST/VALUES return type via PPLReturnTypes.STRING_ARRAY which
   forces ARRAY<VARCHAR> regardless of input element type. Fix: repair the
   PARTIAL aggCall return type to ARRAY<actual-arg0> in OpenSearchAggregateSplitRule
   (PARTIAL only — FINAL keeps the original list to satisfy Volcano's parent
   row-type check on transformTo). The corrected type propagates through
   PARTIAL output → StageInputScan → FINAL substrait base_schema.

2. DataFusion's substrait consumer ignores AggregationPhase, so a FINAL
   array_agg(state) gets lowered as a single-pass aggregate that re-wraps each
   shard's list rather than concatenating. Fix: custom Rust UDAFs `list_merge`
   and `list_merge_distinct` whose `update_batch` un-nests each List<elem> row
   into elements (with optional dedup). Same pattern as TAKE's accumulator.
   The convertor's pre-emit RelShuttle detects the FINAL form (arg0 already
   a list type) and routes LIST → list_merge, VALUES → list_merge_distinct.

PARTIAL still uses DataFusion's native array_agg (with INVOCATION_DISTINCT for
VALUES). Cross-shard testListAcrossShards / testValuesAcrossShards pass.

Signed-off-by: Sandesh Kumar <sandeshkr419@gmail.com>

* analytics-backend-datafusion: relocate state-expanding aggregate rewrites

Addresses two review comments on the state-expanding aggregate (TAKE / FIRST
/ LAST / LIST / VALUES) wiring.

1. Move literal-arg inlining off the post-emit Substrait IR pass and onto
   the isthmus binding layer. Calcite's RelBuilder.aggregate auto-projects
   literal aggregate-args (e.g. TAKE's N) as separate columns and
   AggregateCall.argList only carries column indices, so without inlining
   our Rust UDAFs only see a column reference at construction time. With
   DataFusion's two-stage execution, the Final accumulator dispatches
   through merge_batch — which receives only state columns, not the
   constant — so the limit can never be resolved on the coordinator.

   The fix overrides AggregateFunctionConverter.convert() in createVisitor()
   (next to the existing getSigs() override): after isthmus binds the
   AggregateFunctionInvocation, replace any FieldReference arg pointing to
   a literal Project column with the converted RexLiteral via the supplied
   Function<RexNode, Expression> — same Substrait literal an inlined arg
   would produce, routed through isthmus's standard literal converter.
   SubstraitPlanRewriter.visit(Aggregate) and its simpleStructOffset helper
   are deleted; the post-emit pass now only handles Filter rewrites and
   the PrecisionTimestampLiteral precision fix.

2. Extract the PPL aggregate-name → LOCAL_*_OP rewrite from
   DataFusionFragmentConvertor into a new package-private
   PplAggregateCallRewriter, matching the UntypedNullPreprocessor /
   DatetimeOutputCastRewriter sibling pattern. The if/else-if chain on
   operator name becomes a switch on toUpperCase(Locale.ROOT) and the
   six-way `==` chain that gates already-rewritten calls becomes a
   Set<SqlAggFunction> contains check. The two call sites in
   DataFusionFragmentConvertor now read like the other preprocessor
   invocations.

CoordinatorReduceIT (18 tests), integTestMemtable, and integTestStreaming
all pass.

Signed-off-by: Sandesh Kumar <sandeshkr419@gmail.com>

---------

Signed-off-by: Sandesh Kumar <sandeshkr419@gmail.com>

Author: sandeshkr419

sandbox/libs/analytics-framework/src/main/java/org/opensearch/analytics/spi/AggregateFunction.java

@@ -9,9 +9,12 @@
 package org.opensearch.analytics.spi;
 
 import org.apache.arrow.vector.types.pojo.ArrowType;
+import org.apache.calcite.rel.type.RelDataType;
+import org.apache.calcite.rel.type.RelDataTypeFactory;
 import org.apache.calcite.sql.SqlAggFunction;
 import org.apache.calcite.sql.SqlKind;
 import org.apache.calcite.sql.fun.SqlStdOperatorTable;
+import org.apache.calcite.sql.type.SqlTypeName;
 
 import java.util.List;
 
@@ -50,10 +53,12 @@ public enum AggregateFunction {
     COLLECT(Type.STATE_EXPANDING, SqlKind.COLLECT),
     LISTAGG(Type.STATE_EXPANDING, SqlKind.LISTAGG),
 
-    // Approximate — probabilistic, fixed-size state. Engine-native merge: null reducer
-    // means the field is reduced by this same function (APPROX_COUNT_DISTINCT merges
-    // partial HLL sketches into a final sketch).
-    APPROX_COUNT_DISTINCT(Type.APPROXIMATE, SqlKind.OTHER, fields(IF("sketch", new ArrowType.Binary(), null)));
+    APPROX_COUNT_DISTINCT(Type.APPROXIMATE, SqlKind.OTHER, fields(IF("sketch", new ArrowType.Binary(), null))),
+    TAKE(Type.STATE_EXPANDING, SqlKind.OTHER, fields(IF("take_state", IntermediateTypeResolver.passThroughArg0(), null))),
+    FIRST(Type.STATE_EXPANDING, SqlKind.OTHER, fields(IF("first_state", IntermediateTypeResolver.passThroughArg0(), null))),
+    LAST(Type.STATE_EXPANDING, SqlKind.OTHER, fields(IF("last_state", IntermediateTypeResolver.passThroughArg0(), null))),
+    LIST(Type.STATE_EXPANDING, SqlKind.OTHER, fields(IF("list_state", IntermediateTypeResolver.passThroughArg0(), null))),
+    VALUES(Type.STATE_EXPANDING, SqlKind.OTHER, fields(IF("values_state", IntermediateTypeResolver.passThroughArg0(), null)));
 
     /** Category of aggregate function. Affects execution strategy (shuffle vs map-reduce). */
     public enum Type {
@@ -63,8 +68,71 @@ public enum Type {
         APPROXIMATE
     }
 
-    /** Describes one intermediate field emitted by a partial aggregate. A null reducer means "self" (the owning enum constant). */
-    public record IntermediateField(String name, ArrowType arrowType, AggregateFunction reducer) {
+    /**
+     * Describes one intermediate field emitted by a partial aggregate. A null reducer means
+     * "self" (the owning enum constant).
+     *
+     * <p>The {@code typeResolver} produces the field's Calcite type given the FINAL aggregate
+     * call's input arg types. For fixed-shape states (HLL sketch is always Binary, COUNT
+     * counter is always Int64) the resolver ignores its input and returns a constant; for
+     * input-parameterised states (e.g. {@code take(field, N)}'s buffer is {@code list<field>})
+     * the resolver derives the shape from arg 0. Construct via
+     * {@link IntermediateTypeResolver#fixed(ArrowType)} and
+     * {@link IntermediateTypeResolver#passThroughArg0()}.
+     */
+    public record IntermediateField(String name, IntermediateTypeResolver typeResolver, AggregateFunction reducer) {
+    }
+
+    /**
+     * Computes the intermediate-field type from an aggregate call's arg types. Implementations
+     * must be pure: same input → same output. Two flavours:
+     * <ul>
+     *   <li><b>Fixed</b> ({@link #fixed(ArrowType)}) — returns a constant Arrow type wrapped
+     *       in the corresponding Calcite type. Used for COUNT (Int64) and APPROX_COUNT_DISTINCT
+     *       (Binary).</li>
+     *   <li><b>Input-parameterised</b> (custom impls like {@link #passThroughArg0()}) — derives
+     *       the type from {@code argTypes}. Used for {@code take}/{@code list}/{@code values}
+     *       whose state shape is {@code list<arg0>}.</li>
+     * </ul>
+     */
+    @FunctionalInterface
+    public interface IntermediateTypeResolver {
+        /** Resolve the intermediate field's Calcite type. */
+        RelDataType resolve(List<RelDataType> argTypes, RelDataTypeFactory typeFactory);
+
+        /** Resolver that always returns the same Arrow type, irrespective of arg types. */
+        static IntermediateTypeResolver fixed(ArrowType arrowType) {
+            return (argTypes, typeFactory) -> ArrowToCalciteTypeMapper.toCalcite(arrowType, typeFactory);
+        }
+
+        /**
+         * Pass arg 0's type through unchanged. Used by state-expanding aggregates whose
+         * FINAL re-aggregates over PARTIAL's output column — the column type already
+         * equals the desired exchange shape.
+         */
+        static IntermediateTypeResolver passThroughArg0() {
+            return (argTypes, typeFactory) -> {
+                if (argTypes.isEmpty()) {
+                    throw new IllegalStateException("passThroughArg0 resolver requires at least one arg type");
+                }
+                return argTypes.get(0);
+            };
+        }
+    }
+
+    /**
+     * Internal Arrow → Calcite mapper used by {@link IntermediateTypeResolver#fixed}. Lives
+     * in the SPI module so {@code IntermediateField} stays self-contained — no dependency
+     * on the planner module. Add cases as new fixed-state shapes appear.
+     */
+    private static final class ArrowToCalciteTypeMapper {
+        static RelDataType toCalcite(ArrowType t, RelDataTypeFactory f) {
+            return switch (t) {
+                case ArrowType.Int i when i.getBitWidth() == 64 -> f.createSqlType(SqlTypeName.BIGINT);
+                case ArrowType.Binary b -> f.createSqlType(SqlTypeName.VARBINARY, Integer.MAX_VALUE);
+                default -> throw new IllegalArgumentException("Unsupported fixed Arrow type for IntermediateField: " + t);
+            };
+        }
     }
 
     private final Type type;
@@ -93,7 +161,7 @@ public SqlKind getSqlKind() {
     public List<IntermediateField> intermediateFields() {
         if (intermediateFields == null) return null;
         return intermediateFields.stream()
-            .map(f -> f.reducer() == null ? new IntermediateField(f.name(), f.arrowType(), this) : f)
+            .map(f -> f.reducer() == null ? new IntermediateField(f.name(), f.typeResolver(), this) : f)
             .toList();
     }
 
@@ -163,6 +231,10 @@ private static List<IntermediateField> fields(IntermediateField... fs) {
     }
 
     private static IntermediateField IF(String name, ArrowType arrowType, AggregateFunction reducer) {
-        return new IntermediateField(name, arrowType, reducer);
+        return new IntermediateField(name, IntermediateTypeResolver.fixed(arrowType), reducer);
+    }
+
+    private static IntermediateField IF(String name, IntermediateTypeResolver typeResolver, AggregateFunction reducer) {
+        return new IntermediateField(name, typeResolver, reducer);
     }
 }

sandbox/libs/analytics-framework/src/test/java/org/opensearch/analytics/spi/AggregateFunctionTests.java

@@ -8,18 +8,26 @@
 
 package org.opensearch.analytics.spi;
 
-import org.apache.arrow.vector.types.pojo.ArrowType;
+import org.apache.calcite.jdbc.JavaTypeFactoryImpl;
+import org.apache.calcite.rel.type.RelDataType;
+import org.apache.calcite.rel.type.RelDataTypeFactory;
 import org.apache.calcite.sql.SqlKind;
+import org.apache.calcite.sql.type.SqlTypeName;
 import org.opensearch.test.OpenSearchTestCase;
 
 import java.util.List;
 
 import static org.opensearch.analytics.spi.AggregateFunction.APPROX_COUNT_DISTINCT;
 import static org.opensearch.analytics.spi.AggregateFunction.AVG;
 import static org.opensearch.analytics.spi.AggregateFunction.COUNT;
+import static org.opensearch.analytics.spi.AggregateFunction.FIRST;
+import static org.opensearch.analytics.spi.AggregateFunction.LAST;
+import static org.opensearch.analytics.spi.AggregateFunction.LIST;
 import static org.opensearch.analytics.spi.AggregateFunction.MAX;
 import static org.opensearch.analytics.spi.AggregateFunction.MIN;
 import static org.opensearch.analytics.spi.AggregateFunction.SUM;
+import static org.opensearch.analytics.spi.AggregateFunction.TAKE;
+import static org.opensearch.analytics.spi.AggregateFunction.VALUES;
 
 /**
  * Asserts the enum carries the right shape per function for the resolver's three
@@ -34,6 +42,13 @@
  */
 public class AggregateFunctionTests extends OpenSearchTestCase {
 
+    private final RelDataTypeFactory typeFactory = new JavaTypeFactoryImpl();
+    private final RelDataType integer = typeFactory.createSqlType(SqlTypeName.INTEGER);
+
+    private RelDataType resolve(AggregateFunction.IntermediateField field, RelDataType arg0) {
+        return field.typeResolver().resolve(List.of(arg0), typeFactory);
+    }
+
     // ── Pass-through: SUM / MIN / MAX ──
 
     public void testSumHasNoDecomposition() {
@@ -52,8 +67,7 @@ public void testCountIntermediateFields() {
         assertEquals(1, fields.size());
         assertEquals("count", fields.get(0).name());
         assertSame(SUM, fields.get(0).reducer());
-        assertTrue(fields.get(0).arrowType() instanceof ArrowType.Int);
-        assertEquals(64, ((ArrowType.Int) fields.get(0).arrowType()).getBitWidth());
+        assertEquals(SqlTypeName.BIGINT, resolve(fields.get(0), integer).getSqlTypeName());
     }
 
     // ── AVG / STDDEV / VAR: handled by Calcite's reduce rule — no enum metadata ──
@@ -77,7 +91,77 @@ public void testApproxCountDistinctReducerIsSelf() {
         assertEquals(1, fields.size());
         assertEquals("sketch", fields.get(0).name());
         assertSame(APPROX_COUNT_DISTINCT, fields.get(0).reducer());
-        assertTrue(fields.get(0).arrowType() instanceof ArrowType.Binary);
+        assertEquals(SqlTypeName.VARBINARY, resolve(fields.get(0), integer).getSqlTypeName());
+    }
+
+    // ── TAKE: engine-native (single field, reducer == self, parameterised resolver) ──
+
+    public void testTakeHasDecomposition() {
+        assertTrue(TAKE.hasDecomposition());
+    }
+
+    public void testTakeReducerIsSelfAndResolverIsParameterised() {
+        List<AggregateFunction.IntermediateField> fields = TAKE.intermediateFields();
+        assertEquals(1, fields.size());
+        assertEquals("take_state", fields.get(0).name());
+        assertSame(TAKE, fields.get(0).reducer());
+        assertEquals("passThroughArg0 returns arg0", integer, resolve(fields.get(0), integer));
+    }
+
+    // ── FIRST: engine-native (single field, reducer == self, parameterised resolver) ──
+
+    public void testFirstHasDecomposition() {
+        assertTrue(FIRST.hasDecomposition());
+    }
+
+    public void testFirstReducerIsSelf() {
+        List<AggregateFunction.IntermediateField> fields = FIRST.intermediateFields();
+        assertEquals(1, fields.size());
+        assertEquals("first_state", fields.get(0).name());
+        assertSame(FIRST, fields.get(0).reducer());
+        assertEquals(integer, resolve(fields.get(0), integer));
+    }
+
+    // ── LAST: engine-native (single field, reducer == self, parameterised resolver) ──
+
+    public void testLastHasDecomposition() {
+        assertTrue(LAST.hasDecomposition());
+    }
+
+    public void testLastReducerIsSelf() {
+        List<AggregateFunction.IntermediateField> fields = LAST.intermediateFields();
+        assertEquals(1, fields.size());
+        assertEquals("last_state", fields.get(0).name());
+        assertSame(LAST, fields.get(0).reducer());
+        assertEquals(integer, resolve(fields.get(0), integer));
+    }
+
+    // ── LIST: engine-native (single field, reducer == self, parameterised resolver) ──
+
+    public void testListHasDecomposition() {
+        assertTrue(LIST.hasDecomposition());
+    }
+
+    public void testListReducerIsSelf() {
+        List<AggregateFunction.IntermediateField> fields = LIST.intermediateFields();
+        assertEquals(1, fields.size());
+        assertEquals("list_state", fields.get(0).name());
+        assertSame(LIST, fields.get(0).reducer());
+        assertEquals(integer, resolve(fields.get(0), integer));
+    }
+
+    // ── VALUES: engine-native (single field, reducer == self, parameterised resolver) ──
+
+    public void testValuesHasDecomposition() {
+        assertTrue(VALUES.hasDecomposition());
+    }
+
+    public void testValuesReducerIsSelf() {
+        List<AggregateFunction.IntermediateField> fields = VALUES.intermediateFields();
+        assertEquals(1, fields.size());
+        assertEquals("values_state", fields.get(0).name());
+        assertSame(VALUES, fields.get(0).reducer());
+        assertEquals(integer, resolve(fields.get(0), integer));
     }
 
     // ── fromSqlKind still works ──

sandbox/plugins/analytics-backend-datafusion/rust/src/api.rs

@@ -534,6 +534,7 @@ pub unsafe fn sql_to_substrait(
             .build();
         let ctx = datafusion::prelude::SessionContext::new_with_state(state);
         crate::udf::register_all(&ctx);
+        crate::udaf::register_all(&ctx);
 
         let listing_options = ListingOptions::new(Arc::new(ParquetFormat::new()))
             .with_file_extension(".parquet")
@@ -585,6 +586,7 @@ fn derive_schema_from_partial_plan(
         .build();
     let ctx = SessionContext::new_with_state(state);
     crate::udf::register_all(&ctx);
+    crate::udaf::register_all(&ctx);
 
     let extensions = Extensions::default();
     let session_state = ctx.state();

sandbox/plugins/analytics-backend-datafusion/rust/src/indexed_executor.rs

@@ -148,6 +148,7 @@ pub async fn execute_indexed_query(
     ctx.register_udf(create_index_filter_udf());
     ctx.register_udf(crate::indexed_table::substrait_to_tree::create_delegation_possible_udf());
     crate::udf::register_all(&ctx);
+    crate::udaf::register_all(&ctx);
 
     // Register default ListingTable so substrait consumer can resolve the table
     let listing_options = datafusion::datasource::listing::ListingOptions::new(

sandbox/plugins/analytics-backend-datafusion/rust/src/lib.rs

@@ -33,6 +33,7 @@ pub mod runtime_manager;
 pub mod schema_coerce;
 pub mod session_context;
 pub mod statistics_cache;
+pub mod udaf;
 pub mod udf;
 pub mod stats;
 pub mod task_monitors;

sandbox/plugins/analytics-backend-datafusion/rust/src/local_executor.rs

@@ -74,6 +74,7 @@ impl LocalSession {
             .build();
         let ctx = SessionContext::new_with_state(state);
         crate::udf::register_all(&ctx);
+        crate::udaf::register_all(&ctx);
         Self { ctx, prepared_plan: None }
     }
 

sandbox/plugins/analytics-backend-datafusion/rust/src/query_executor.rs

@@ -113,6 +113,7 @@ pub async fn execute_query(
 
     let ctx = SessionContext::new_with_state(state);
     crate::udf::register_all(&ctx);
+    crate::udaf::register_all(&ctx);
 
     // Register table via ListingTable — all IO goes through object store
     let file_format = ParquetFormat::new();

sandbox/plugins/analytics-backend-datafusion/rust/src/session_context.rs

@@ -133,6 +133,7 @@ pub async unsafe fn create_session_context(
     // Without this, fragment execution fails with "Unsupported function name" because
     // df_execute_with_context reuses this handle's ctx instead of building a fresh one.
     crate::udf::register_all(&ctx);
+    crate::udaf::register_all(&ctx);
 
     // Register default ListingTable for parquet scans.
     let listing_options = ListingOptions::new(Arc::new(ParquetFormat::default()))