paimon_datafusion/table/
mod.rs

1// Licensed to the Apache Software Foundation (ASF) under one
2// or more contributor license agreements.  See the NOTICE file
3// distributed with this work for additional information
4// regarding copyright ownership.  The ASF licenses this file
5// to you under the Apache License, Version 2.0 (the
6// "License"); you may not use this file except in compliance
7// with the License.  You may obtain a copy of the License at
8//
9//   http://www.apache.org/licenses/LICENSE-2.0
10//
11// Unless required by applicable law or agreed to in writing,
12// software distributed under the License is distributed on an
13// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14// KIND, either express or implied.  See the License for the
15// specific language governing permissions and limitations
16// under the License.
17
18//! Paimon table provider for DataFusion.
19
20use std::any::Any;
21use std::sync::Arc;
22
23use async_trait::async_trait;
24use datafusion::arrow::datatypes::{Field, Schema, SchemaRef as ArrowSchemaRef};
25use datafusion::catalog::Session;
26use datafusion::datasource::sink::DataSinkExec;
27use datafusion::datasource::{TableProvider, TableType};
28use datafusion::error::Result as DFResult;
29use datafusion::logical_expr::dml::InsertOp;
30use datafusion::logical_expr::{Expr, TableProviderFilterPushDown};
31use datafusion::physical_plan::ExecutionPlan;
32use paimon::table::Table;
33
34use crate::physical_plan::PaimonDataSink;
35
36use crate::error::to_datafusion_error;
37#[cfg(test)]
38use crate::filter_pushdown::build_pushed_predicate;
39use crate::filter_pushdown::{analyze_filters, classify_filter_pushdown};
40use crate::physical_plan::PaimonTableScan;
41use crate::runtime::await_with_runtime;
42
43/// Read-only table provider for a Paimon table.
44///
45/// Supports full table scan, column projection, and predicate pushdown for
46/// planning. Partition predicates prune splits eagerly, while supported
47/// non-partition data predicates may also be reused by the Parquet read path
48/// for row-group pruning and partial decode-time filtering.
49///
50/// DataFusion still treats pushed filters as inexact because unsupported
51/// predicates and non-Parquet reads remain residual filters.
52#[derive(Debug, Clone)]
53pub struct PaimonTableProvider {
54    table: Table,
55    schema: ArrowSchemaRef,
56}
57
58impl PaimonTableProvider {
59    /// Create a table provider from a Paimon table.
60    ///
61    /// Loads the table schema and converts it to Arrow for DataFusion.
62    pub fn try_new(table: Table) -> DFResult<Self> {
63        let mut fields = table.schema().fields().to_vec();
64        let core_options = paimon::spec::CoreOptions::new(table.schema().options());
65        if core_options.data_evolution_enabled() {
66            fields.push(paimon::spec::DataField::new(
67                paimon::spec::ROW_ID_FIELD_ID,
68                paimon::spec::ROW_ID_FIELD_NAME.to_string(),
69                paimon::spec::DataType::BigInt(paimon::spec::BigIntType::with_nullable(true)),
70            ));
71        }
72        let schema =
73            paimon::arrow::build_target_arrow_schema(&fields).map_err(to_datafusion_error)?;
74        Ok(Self { table, schema })
75    }
76
77    pub fn table(&self) -> &Table {
78        &self.table
79    }
80}
81
82/// Distribute `items` into `num_buckets` groups using round-robin assignment.
83pub(crate) fn bucket_round_robin<T>(items: Vec<T>, num_buckets: usize) -> Vec<Vec<T>> {
84    let mut buckets: Vec<Vec<T>> = (0..num_buckets).map(|_| Vec::new()).collect();
85    for (index, item) in items.into_iter().enumerate() {
86        buckets[index % num_buckets].push(item);
87    }
88    buckets
89}
90
91/// Build parameters for [`PaimonTableScan`].
92pub(crate) struct PaimonScanBuilder<'a> {
93    pub(crate) table: &'a Table,
94    pub(crate) schema: &'a ArrowSchemaRef,
95    pub(crate) plan: &'a paimon::table::Plan,
96    pub(crate) projection: Option<&'a Vec<usize>>,
97    pub(crate) pushed_predicate: Option<paimon::spec::Predicate>,
98    pub(crate) limit: Option<usize>,
99    pub(crate) target_partitions: usize,
100    pub(crate) filter_exact: bool,
101}
102
103impl PaimonScanBuilder<'_> {
104    /// Build a [`PaimonTableScan`] from the configured parameters.
105    pub(crate) fn build(self) -> DFResult<Arc<dyn ExecutionPlan>> {
106        let (projected_schema, projected_columns) = if let Some(indices) = self.projection {
107            let fields: Vec<Field> = indices
108                .iter()
109                .map(|&i| self.schema.field(i).clone())
110                .collect();
111            let column_names: Vec<String> = fields.iter().map(|f| f.name().clone()).collect();
112            (Arc::new(Schema::new(fields)), Some(column_names))
113        } else {
114            let column_names: Vec<String> = self
115                .schema
116                .fields()
117                .iter()
118                .map(|f| f.name().clone())
119                .collect();
120            (self.schema.clone(), Some(column_names))
121        };
122
123        let splits = self.plan.splits().to_vec();
124        let planned_partitions: Vec<Arc<[_]>> = if splits.is_empty() {
125            vec![Arc::from(Vec::new())]
126        } else {
127            let num_partitions = splits.len().min(self.target_partitions.max(1));
128            bucket_round_robin(splits, num_partitions)
129                .into_iter()
130                .map(Arc::from)
131                .collect()
132        };
133
134        Ok(Arc::new(PaimonTableScan::new(
135            projected_schema,
136            self.table.clone(),
137            projected_columns,
138            self.pushed_predicate,
139            planned_partitions,
140            self.limit,
141            self.filter_exact,
142        )))
143    }
144}
145
146#[async_trait]
147impl TableProvider for PaimonTableProvider {
148    fn as_any(&self) -> &dyn Any {
149        self
150    }
151
152    fn schema(&self) -> ArrowSchemaRef {
153        self.schema.clone()
154    }
155
156    fn table_type(&self) -> TableType {
157        TableType::Base
158    }
159
160    async fn scan(
161        &self,
162        state: &dyn Session,
163        projection: Option<&Vec<usize>>,
164        filters: &[Expr],
165        limit: Option<usize>,
166    ) -> DFResult<Arc<dyn ExecutionPlan>> {
167        // Plan splits eagerly so we know partition count upfront.
168        let filter_analysis = analyze_filters(filters, self.table.schema().fields());
169        let mut read_builder = self.table.new_read_builder();
170        if let Some(filter) = filter_analysis.pushed_predicate.clone() {
171            read_builder.with_filter(filter);
172        }
173        let pushed_limit = limit.filter(|_| !filter_analysis.has_untranslated_residual);
174        if let Some(limit) = pushed_limit {
175            read_builder.with_limit(limit);
176        }
177        let scan = read_builder.new_scan();
178        // DataFusion's Python FFI may poll `TableProvider::scan()` without an active
179        // Tokio runtime. `scan.plan()` can reach OpenDAL/Tokio filesystem calls while
180        // reading Paimon metadata, so we must provide a runtime here instead of
181        // assuming the caller already entered one.
182        let plan = await_with_runtime(scan.plan())
183            .await
184            .map_err(to_datafusion_error)?;
185
186        let target = state.config_options().execution.target_partitions;
187        let filter_exact = !filter_analysis.has_untranslated_residual
188            && filter_analysis
189                .pushed_predicate
190                .as_ref()
191                .is_none_or(|p| read_builder.is_exact_filter_pushdown(p));
192        PaimonScanBuilder {
193            table: &self.table,
194            schema: &self.schema,
195            plan: &plan,
196            projection,
197            pushed_predicate: filter_analysis.pushed_predicate,
198            limit: pushed_limit,
199            target_partitions: target,
200            filter_exact,
201        }
202        .build()
203    }
204
205    async fn insert_into(
206        &self,
207        _state: &dyn Session,
208        input: Arc<dyn ExecutionPlan>,
209        insert_op: InsertOp,
210    ) -> DFResult<Arc<dyn ExecutionPlan>> {
211        let overwrite = match insert_op {
212            InsertOp::Append => false,
213            InsertOp::Overwrite => true,
214            other => {
215                return Err(datafusion::error::DataFusionError::NotImplemented(format!(
216                    "{other} is not supported for Paimon tables"
217                )));
218            }
219        };
220        let sink = PaimonDataSink::new(self.table.clone(), self.schema.clone(), overwrite);
221        Ok(Arc::new(DataSinkExec::new(input, Arc::new(sink), None)))
222    }
223
224    fn supports_filters_pushdown(
225        &self,
226        filters: &[&Expr],
227    ) -> DFResult<Vec<TableProviderFilterPushDown>> {
228        let fields = self.table.schema().fields();
229        let read_builder = self.table.new_read_builder();
230
231        Ok(filters
232            .iter()
233            .map(|filter| {
234                classify_filter_pushdown(filter, fields, |predicate| {
235                    read_builder.is_exact_filter_pushdown(predicate)
236                })
237            })
238            .collect())
239    }
240}
241
242#[cfg(test)]
243mod tests {
244    use super::*;
245    use std::collections::BTreeSet;
246    use std::sync::Arc;
247
248    use datafusion::datasource::TableProvider;
249    use datafusion::logical_expr::{col, lit, Expr};
250    use datafusion::prelude::{SessionConfig, SessionContext};
251    use paimon::catalog::Identifier;
252    use paimon::{Catalog, CatalogOptions, DataSplit, FileSystemCatalog, Options};
253
254    use crate::physical_plan::PaimonTableScan;
255
256    #[test]
257    fn test_bucket_round_robin_distributes_evenly() {
258        let result = bucket_round_robin(vec![0, 1, 2, 3, 4], 3);
259        assert_eq!(result, vec![vec![0, 3], vec![1, 4], vec![2]]);
260    }
261
262    #[test]
263    fn test_bucket_round_robin_fewer_items_than_buckets() {
264        let result = bucket_round_robin(vec![10, 20], 2);
265        assert_eq!(result, vec![vec![10], vec![20]]);
266    }
267
268    #[test]
269    fn test_bucket_round_robin_single_bucket() {
270        let result = bucket_round_robin(vec![1, 2, 3], 1);
271        assert_eq!(result, vec![vec![1, 2, 3]]);
272    }
273
274    fn get_test_warehouse() -> String {
275        std::env::var("PAIMON_TEST_WAREHOUSE")
276            .unwrap_or_else(|_| "/tmp/paimon-warehouse".to_string())
277    }
278
279    fn create_catalog() -> FileSystemCatalog {
280        let warehouse = get_test_warehouse();
281        let mut options = Options::new();
282        options.set(CatalogOptions::WAREHOUSE, warehouse);
283        FileSystemCatalog::new(options).expect("Failed to create catalog")
284    }
285
286    async fn create_provider(table_name: &str) -> PaimonTableProvider {
287        let catalog = create_catalog();
288        let identifier = Identifier::new("default", table_name);
289        let table = catalog
290            .get_table(&identifier)
291            .await
292            .expect("Failed to get table");
293
294        PaimonTableProvider::try_new(table).expect("Failed to create table provider")
295    }
296
297    async fn plan_partitions(
298        provider: &PaimonTableProvider,
299        filters: Vec<Expr>,
300        limit: Option<usize>,
301    ) -> Vec<Arc<[DataSplit]>> {
302        let plan = plan_scan(provider, filters, limit).await;
303        let scan = plan
304            .as_any()
305            .downcast_ref::<PaimonTableScan>()
306            .expect("Expected PaimonTableScan");
307
308        scan.planned_partitions().to_vec()
309    }
310
311    async fn plan_scan(
312        provider: &PaimonTableProvider,
313        filters: Vec<Expr>,
314        limit: Option<usize>,
315    ) -> Arc<dyn ExecutionPlan> {
316        let config = SessionConfig::new().with_target_partitions(8);
317        let ctx = SessionContext::new_with_config(config);
318        let state = ctx.state();
319        provider
320            .scan(&state, None, &filters, limit)
321            .await
322            .expect("scan() should succeed")
323    }
324
325    fn extract_dt_partition_set(planned_partitions: &[Arc<[DataSplit]>]) -> BTreeSet<String> {
326        planned_partitions
327            .iter()
328            .flat_map(|splits| splits.iter())
329            .map(|split| {
330                split
331                    .partition()
332                    .get_string(0)
333                    .expect("Failed to decode dt")
334                    .to_string()
335            })
336            .collect()
337    }
338
339    fn extract_dt_hr_partition_set(
340        planned_partitions: &[Arc<[DataSplit]>],
341    ) -> BTreeSet<(String, i32)> {
342        planned_partitions
343            .iter()
344            .flat_map(|splits| splits.iter())
345            .map(|split| {
346                let partition = split.partition();
347                (
348                    partition
349                        .get_string(0)
350                        .expect("Failed to decode dt")
351                        .to_string(),
352                    partition.get_int(1).expect("Failed to decode hr"),
353                )
354            })
355            .collect()
356    }
357
358    #[tokio::test]
359    async fn test_scan_partition_filter_plans_matching_partition_set() {
360        let provider = create_provider("partitioned_log_table").await;
361        let planned_partitions =
362            plan_partitions(&provider, vec![col("dt").eq(lit("2024-01-01"))], None).await;
363
364        assert_eq!(
365            extract_dt_partition_set(&planned_partitions),
366            BTreeSet::from(["2024-01-01".to_string()]),
367        );
368    }
369
370    #[tokio::test]
371    async fn test_scan_mixed_and_filter_keeps_partition_pruning() {
372        let provider = create_provider("partitioned_log_table").await;
373        let planned_partitions = plan_partitions(
374            &provider,
375            vec![col("dt").eq(lit("2024-01-01")).and(col("id").gt(lit(1)))],
376            None,
377        )
378        .await;
379
380        assert_eq!(
381            extract_dt_partition_set(&planned_partitions),
382            BTreeSet::from(["2024-01-01".to_string()]),
383        );
384    }
385
386    #[tokio::test]
387    async fn test_scan_multi_partition_filter_plans_exact_partition_set() {
388        let provider = create_provider("multi_partitioned_log_table").await;
389
390        let dt_only_partitions =
391            plan_partitions(&provider, vec![col("dt").eq(lit("2024-01-01"))], None).await;
392        let dt_hr_partitions = plan_partitions(
393            &provider,
394            vec![col("dt").eq(lit("2024-01-01")).and(col("hr").eq(lit(10)))],
395            None,
396        )
397        .await;
398
399        assert_eq!(
400            extract_dt_hr_partition_set(&dt_only_partitions),
401            BTreeSet::from([
402                ("2024-01-01".to_string(), 10),
403                ("2024-01-01".to_string(), 20),
404            ]),
405        );
406        assert_eq!(
407            extract_dt_hr_partition_set(&dt_hr_partitions),
408            BTreeSet::from([("2024-01-01".to_string(), 10)]),
409        );
410    }
411
412    #[tokio::test]
413    async fn test_scan_partially_translated_filter_keeps_partition_pruning_but_skips_limit_hint() {
414        let provider = create_provider("multi_partitioned_log_table").await;
415        let filter = col("dt")
416            .eq(lit("2024-01-01"))
417            .and(Expr::Not(Box::new(col("hr").eq(lit(10)))));
418        let full_plan = plan_partitions(&provider, vec![filter.clone()], None).await;
419        let plan = plan_scan(&provider, vec![filter], Some(1)).await;
420        let scan = plan
421            .as_any()
422            .downcast_ref::<PaimonTableScan>()
423            .expect("Expected PaimonTableScan");
424
425        assert_eq!(scan.limit(), None);
426        assert_eq!(
427            extract_dt_hr_partition_set(scan.planned_partitions()),
428            BTreeSet::from([
429                ("2024-01-01".to_string(), 10),
430                ("2024-01-01".to_string(), 20),
431            ]),
432        );
433        assert_eq!(
434            scan.planned_partitions()
435                .iter()
436                .map(|partition| partition.len())
437                .sum::<usize>(),
438            full_plan
439                .iter()
440                .map(|partition| partition.len())
441                .sum::<usize>()
442        );
443    }
444
445    #[tokio::test]
446    async fn test_scan_keeps_pushed_predicate_for_execute() {
447        let provider = create_provider("partitioned_log_table").await;
448        let filter = col("id").gt(lit(1));
449
450        let config = SessionConfig::new().with_target_partitions(8);
451        let ctx = SessionContext::new_with_config(config);
452        let state = ctx.state();
453        let plan = provider
454            .scan(&state, None, std::slice::from_ref(&filter), None)
455            .await
456            .expect("scan() should succeed");
457        let scan = plan
458            .as_any()
459            .downcast_ref::<PaimonTableScan>()
460            .expect("Expected PaimonTableScan");
461
462        let expected = build_pushed_predicate(&[filter], provider.table().schema().fields())
463            .expect("data filter should translate");
464
465        assert_eq!(scan.pushed_predicate(), Some(&expected));
466    }
467
468    #[tokio::test]
469    async fn test_scan_applies_limit_hint_only_when_safe() {
470        let provider = create_provider("partitioned_log_table").await;
471        let full_plan = plan_partitions(&provider, vec![], None).await;
472        let plan = plan_scan(&provider, vec![], Some(1)).await;
473        let scan = plan
474            .as_any()
475            .downcast_ref::<PaimonTableScan>()
476            .expect("Expected PaimonTableScan");
477
478        assert_eq!(scan.limit(), Some(1));
479        assert!(
480            scan.planned_partitions()
481                .iter()
482                .map(|partition| partition.len())
483                .sum::<usize>()
484                < full_plan
485                    .iter()
486                    .map(|partition| partition.len())
487                    .sum::<usize>()
488        );
489    }
490
491    #[tokio::test]
492    async fn test_scan_keeps_limit_but_skips_limit_pruning_for_data_filters() {
493        let provider = create_provider("partitioned_log_table").await;
494        let filter = col("id").gt(lit(1));
495        let full_plan = plan_partitions(&provider, vec![filter.clone()], None).await;
496        let plan = plan_scan(&provider, vec![filter], Some(1)).await;
497        let scan = plan
498            .as_any()
499            .downcast_ref::<PaimonTableScan>()
500            .expect("Expected PaimonTableScan");
501
502        assert_eq!(scan.limit(), Some(1));
503        assert_eq!(
504            scan.planned_partitions()
505                .iter()
506                .map(|partition| partition.len())
507                .sum::<usize>(),
508            full_plan
509                .iter()
510                .map(|partition| partition.len())
511                .sum::<usize>()
512        );
513    }
514
515    #[tokio::test]
516    async fn test_insert_into_and_read_back() {
517        use paimon::io::FileIOBuilder;
518        use paimon::spec::{DataType, IntType, Schema as PaimonSchema, TableSchema};
519
520        let file_io = FileIOBuilder::new("memory").build().unwrap();
521        let table_path = "memory:/test_df_insert_into";
522        file_io
523            .mkdirs(&format!("{table_path}/snapshot/"))
524            .await
525            .unwrap();
526        file_io
527            .mkdirs(&format!("{table_path}/manifest/"))
528            .await
529            .unwrap();
530
531        let schema = PaimonSchema::builder()
532            .column("id", DataType::Int(IntType::new()))
533            .column("value", DataType::Int(IntType::new()))
534            .build()
535            .unwrap();
536        let table_schema = TableSchema::new(0, &schema);
537        let table = paimon::table::Table::new(
538            file_io,
539            Identifier::new("default", "test_insert"),
540            table_path.to_string(),
541            table_schema,
542            None,
543        );
544
545        let provider = PaimonTableProvider::try_new(table).unwrap();
546        let ctx = SessionContext::new();
547        ctx.register_table("t", Arc::new(provider)).unwrap();
548
549        // INSERT INTO
550        let result = ctx
551            .sql("INSERT INTO t VALUES (1, 10), (2, 20), (3, 30)")
552            .await
553            .unwrap()
554            .collect()
555            .await
556            .unwrap();
557
558        // Verify count output
559        let count_array = result[0]
560            .column(0)
561            .as_any()
562            .downcast_ref::<datafusion::arrow::array::UInt64Array>()
563            .unwrap();
564        assert_eq!(count_array.value(0), 3);
565
566        // Read back
567        let batches = ctx
568            .sql("SELECT id, value FROM t ORDER BY id")
569            .await
570            .unwrap()
571            .collect()
572            .await
573            .unwrap();
574
575        let mut rows = Vec::new();
576        for batch in &batches {
577            let ids = batch
578                .column(0)
579                .as_any()
580                .downcast_ref::<datafusion::arrow::array::Int32Array>()
581                .unwrap();
582            let vals = batch
583                .column(1)
584                .as_any()
585                .downcast_ref::<datafusion::arrow::array::Int32Array>()
586                .unwrap();
587            for i in 0..batch.num_rows() {
588                rows.push((ids.value(i), vals.value(i)));
589            }
590        }
591        assert_eq!(rows, vec![(1, 10), (2, 20), (3, 30)]);
592    }
593
594    #[tokio::test]
595    async fn test_insert_overwrite() {
596        use paimon::io::FileIOBuilder;
597        use paimon::spec::{DataType, IntType, Schema as PaimonSchema, TableSchema, VarCharType};
598
599        let file_io = FileIOBuilder::new("memory").build().unwrap();
600        let table_path = "memory:/test_df_insert_overwrite";
601        file_io
602            .mkdirs(&format!("{table_path}/snapshot/"))
603            .await
604            .unwrap();
605        file_io
606            .mkdirs(&format!("{table_path}/manifest/"))
607            .await
608            .unwrap();
609
610        let schema = PaimonSchema::builder()
611            .column("pt", DataType::VarChar(VarCharType::string_type()))
612            .column("id", DataType::Int(IntType::new()))
613            .partition_keys(["pt"])
614            .build()
615            .unwrap();
616        let table_schema = TableSchema::new(0, &schema);
617        let table = paimon::table::Table::new(
618            file_io,
619            Identifier::new("default", "test_overwrite"),
620            table_path.to_string(),
621            table_schema,
622            None,
623        );
624
625        let provider = PaimonTableProvider::try_new(table).unwrap();
626        let ctx = SessionContext::new();
627        ctx.register_table("t", Arc::new(provider)).unwrap();
628
629        // Initial INSERT: partition "a" and "b"
630        ctx.sql("INSERT INTO t VALUES ('a', 1), ('a', 2), ('b', 3), ('b', 4)")
631            .await
632            .unwrap()
633            .collect()
634            .await
635            .unwrap();
636
637        // INSERT OVERWRITE with only partition "a" data
638        // Should overwrite partition "a" but leave partition "b" intact
639        ctx.sql("INSERT OVERWRITE t VALUES ('a', 10), ('a', 20)")
640            .await
641            .unwrap()
642            .collect()
643            .await
644            .unwrap();
645
646        // Read back
647        let batches = ctx
648            .sql("SELECT pt, id FROM t ORDER BY pt, id")
649            .await
650            .unwrap()
651            .collect()
652            .await
653            .unwrap();
654
655        let mut rows = Vec::new();
656        for batch in &batches {
657            let pts = batch
658                .column(0)
659                .as_any()
660                .downcast_ref::<datafusion::arrow::array::StringArray>()
661                .unwrap();
662            let ids = batch
663                .column(1)
664                .as_any()
665                .downcast_ref::<datafusion::arrow::array::Int32Array>()
666                .unwrap();
667            for i in 0..batch.num_rows() {
668                rows.push((pts.value(i).to_string(), ids.value(i)));
669            }
670        }
671        // Partition "a" overwritten with new data, partition "b" untouched
672        assert_eq!(
673            rows,
674            vec![
675                ("a".to_string(), 10),
676                ("a".to_string(), 20),
677                ("b".to_string(), 3),
678                ("b".to_string(), 4),
679            ]
680        );
681    }
682
683    #[tokio::test]
684    async fn test_insert_overwrite_unpartitioned() {
685        use paimon::io::FileIOBuilder;
686        use paimon::spec::{DataType, IntType, Schema as PaimonSchema, TableSchema};
687
688        let file_io = FileIOBuilder::new("memory").build().unwrap();
689        let table_path = "memory:/test_df_insert_overwrite_unpart";
690        file_io
691            .mkdirs(&format!("{table_path}/snapshot/"))
692            .await
693            .unwrap();
694        file_io
695            .mkdirs(&format!("{table_path}/manifest/"))
696            .await
697            .unwrap();
698
699        let schema = PaimonSchema::builder()
700            .column("id", DataType::Int(IntType::new()))
701            .column("value", DataType::Int(IntType::new()))
702            .build()
703            .unwrap();
704        let table_schema = TableSchema::new(0, &schema);
705        let table = paimon::table::Table::new(
706            file_io,
707            Identifier::new("default", "test_overwrite_unpart"),
708            table_path.to_string(),
709            table_schema,
710            None,
711        );
712
713        let provider = PaimonTableProvider::try_new(table).unwrap();
714        let ctx = SessionContext::new();
715        ctx.register_table("t", Arc::new(provider)).unwrap();
716
717        // Initial INSERT
718        ctx.sql("INSERT INTO t VALUES (1, 10), (2, 20), (3, 30)")
719            .await
720            .unwrap()
721            .collect()
722            .await
723            .unwrap();
724
725        // INSERT OVERWRITE on unpartitioned table — full table overwrite
726        ctx.sql("INSERT OVERWRITE t VALUES (4, 40), (5, 50)")
727            .await
728            .unwrap()
729            .collect()
730            .await
731            .unwrap();
732
733        let batches = ctx
734            .sql("SELECT id, value FROM t ORDER BY id")
735            .await
736            .unwrap()
737            .collect()
738            .await
739            .unwrap();
740
741        let mut rows = Vec::new();
742        for batch in &batches {
743            let ids = batch
744                .column(0)
745                .as_any()
746                .downcast_ref::<datafusion::arrow::array::Int32Array>()
747                .unwrap();
748            let vals = batch
749                .column(1)
750                .as_any()
751                .downcast_ref::<datafusion::arrow::array::Int32Array>()
752                .unwrap();
753            for i in 0..batch.num_rows() {
754                rows.push((ids.value(i), vals.value(i)));
755            }
756        }
757        // Old data fully replaced
758        assert_eq!(rows, vec![(4, 40), (5, 50)]);
759    }
760}
paimon_datafusion/table/mod.rs

paimon_datafusion/table/
mod.rs
