This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/paimon-rust.git
The following commit(s) were added to refs/heads/main by this push:
new 60101154 test: replace Spark provisioning with golden fixtures (#1234)
60101154 is described below
commit 60101154291fb8d1f7773b6fe720c14fe25f6e65
Author: Jingsong Lee <[email protected]>
AuthorDate: Tue Oct 27 21:20:41 2026 +0800
test: replace Spark provisioning with golden fixtures (#1234)
---
.github/workflows/ci.yml | 14 -
CONTRIBUTING.md | 12 +-
Cargo.lock | 3 +
Makefile | 38 -
bindings/go/Makefile | 6 +-
bindings/go/tests/paimon_test.go | 25 +-
bindings/go/tests/resource_test.go | 5 -
bindings/go/tests/warehouse_test.go | 81 ++
bindings/python/README.md | 6 +-
bindings/python/tests/test_datafusion.py | 61 +-
crates/integration_tests/Cargo.toml | 3 +
crates/integration_tests/tests/read_tables.rs | 288 +----
crates/integrations/datafusion/src/lib.rs | 3 +
crates/integrations/datafusion/src/table/mod.rs | 37 +-
.../datafusion/testdata/warehouse/README.md | 64 +
.../datafusion/testdata/warehouse/mod.rs | 125 ++
.../datafusion/testdata/warehouse/warehouse.tar.gz | Bin 0 -> 118891 bytes
.../datafusion/tests/dynamic_bucket_tables.rs | 36 +-
.../integrations/datafusion/tests/read_tables.rs | 97 +-
dev/docker-compose.yaml | 41 -
dev/spark/Dockerfile | 53 -
dev/spark/README.md | 60 -
dev/spark/provision.py | 1363 --------------------
dev/spark/spark-defaults.conf | 25 -
docs/src/contributing.md | 9 +-
docs/src/getting-started.md | 4 +-
26 files changed, 486 insertions(+), 1973 deletions(-)
diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index 9d9e69d0..feeed88c 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -226,22 +226,16 @@ jobs:
matrix:
include:
- suite: rust
- needs_docker: true
needs_uv: false
- suite: datafusion
- needs_docker: true
needs_uv: false
- suite: python
- needs_docker: true
needs_uv: true
- suite: go
- needs_docker: true
needs_uv: false
- suite: jindo
- needs_docker: false
needs_uv: false
- suite: oss-cpp
- needs_docker: false
needs_uv: false
env:
RUST_LOG: DEBUG
@@ -260,10 +254,6 @@ jobs:
restore-keys: |
integration-${{ runner.os }}-
- - name: Start Docker containers
- if: matrix.needs_docker == true
- run: make docker-up
-
- name: Rust Integration Test
if: matrix.suite == 'rust'
run: cargo test --locked -p paimon-integration-tests --all-targets
@@ -314,10 +304,6 @@ jobs:
io::storage_jindo::tests::test_real_sdk_io_and_failure_mapping
-- --ignored --exact
- - name: Stop Docker containers
- if: ${{ always() && matrix.needs_docker }}
- run: make docker-down
-
- name: Checkout OSS C++ SDK
if: matrix.suite == 'oss-cpp'
uses: actions/checkout@v7
diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md
index 07bf247b..51ec7376 100644
--- a/CONTRIBUTING.md
+++ b/CONTRIBUTING.md
@@ -31,19 +31,17 @@ commands CI runs, which are the ones to reproduce locally:
- `cargo test -p paimon --all-targets --features fulltext,vortex`: Run the
unit tests. Prefer this over a bare `cargo test`, which also builds the Python
binding and can fail to link `libpython` on some platforms.
- `cargo test -p paimon-rest-server --all-targets`: Run the REST server tests.
-The integration suites additionally need a Paimon warehouse written by Spark,
so
-they are not pure `cargo`:
+The integration suites automatically extract the committed Java-written golden
+warehouse into temporary directories. No Spark or Docker setup is needed:
```bash
-make docker-up # builds the Spark image and writes /tmp/paimon-warehouse
cargo test -p paimon-integration-tests --all-targets
cargo test -p paimon-datafusion --all-targets
-make docker-down # tear down when finished
```
-Without that warehouse those tests fail with `TableNotExist` rather than being
-skipped. Override the location with `PAIMON_TEST_WAREHOUSE` if you need a
-different path.
+Set `PAIMON_TEST_WAREHOUSE` to use an existing warehouse instead. Missing
tables
+fail the tests rather than being skipped or generated by the Rust writer. See
+[the golden
warehouse](crates/integrations/datafusion/testdata/warehouse/README.md) for
fixture provenance.
### Setting up the Development Environment
1. Install Rust using `rustup`. Follow the instructions on the [rustup
website](https://rustup.rs/) to install Rust on your system.
diff --git a/Cargo.lock b/Cargo.lock
index b22c9933..a2fb5f15 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -4807,9 +4807,12 @@ dependencies = [
"arrow-array",
"arrow-schema",
"axum",
+ "flate2",
"futures",
"paimon",
"serde_json",
+ "tar",
+ "tempfile",
"tokio",
]
diff --git a/Makefile b/Makefile
deleted file mode 100644
index 1e18aa7f..00000000
--- a/Makefile
+++ /dev/null
@@ -1,38 +0,0 @@
-# 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.
-
-# Version defaults (can be overridden via environment variables)
-PAIMON_VERSION ?= 1.3.1
-
-# Warehouse path for integration tests
-PAIMON_WAREHOUSE ?= /tmp/paimon-warehouse
-
-.EXPORT_ALL_VARIABLES:
-
-# Docker targets for integration tests.
-# Build and run Spark + Paimon container to provision test data.
-# Usage:
-# make docker-up # use default versions
-# PAIMON_VERSION=1.3.0 make docker-up # override Paimon version
-docker-up:
- docker compose -f dev/docker-compose.yaml build \
- --build-arg PAIMON_VERSION=$(PAIMON_VERSION)
- docker compose -f dev/docker-compose.yaml run --rm spark-paimon
-
-# Stop and remove Docker containers, networks, and volumes.
-docker-down:
- docker compose -f dev/docker-compose.yaml down -v
diff --git a/bindings/go/Makefile b/bindings/go/Makefile
index 44ece884..55e36908 100644
--- a/bindings/go/Makefile
+++ b/bindings/go/Makefile
@@ -49,12 +49,10 @@ build:
zstd -19 -f $(TARGET_DIR)/release/libpaimon_c.$(LIB_EXT) -o
$(GO_DIR)/$(LIB_NAME).zst
# Run Go integration tests.
-# Requires test data: run 'make docker-up' from the repo root first.
+# The tests extract the committed golden warehouse into temporary directories.
# CGO is needed by arrow-go's cdata package (imported by the main paimon
package).
-PAIMON_TEST_WAREHOUSE ?= /tmp/paimon-warehouse
-
test: build
- cd $(GO_DIR)/tests && PAIMON_TEST_WAREHOUSE=$(PAIMON_TEST_WAREHOUSE) go
test -v ./...
+ cd $(GO_DIR)/tests && go test -v ./...
clean:
cd $(RUST_ROOT) && cargo clean
diff --git a/bindings/go/tests/paimon_test.go b/bindings/go/tests/paimon_test.go
index 8a359b0c..bbb257db 100644
--- a/bindings/go/tests/paimon_test.go
+++ b/bindings/go/tests/paimon_test.go
@@ -46,14 +46,6 @@ type partitionedRow struct {
dt string
}
-func testWarehouse() string {
- warehouse := os.Getenv("PAIMON_TEST_WAREHOUSE")
- if warehouse == "" {
- return "/tmp/paimon-warehouse"
- }
- return warehouse
-}
-
func copyDirectory(source, target string) error {
info, err := os.Stat(source)
if err != nil {
@@ -132,11 +124,8 @@ func openCopiedTestTable(t *testing.T) *paimon.Table {
func openCopiedTable(t *testing.T, tableName string) *paimon.Table {
t.Helper()
- warehouse := testWarehouse()
+ warehouse := testWarehouse(t)
source := filepath.Join(warehouse, "default.db", tableName)
- if _, err := os.Stat(source); os.IsNotExist(err) {
- t.Skipf("Skipping: table %s does not exist (run 'make
docker-up' first)", source)
- }
targetWarehouse := t.TempDir()
target := filepath.Join(targetWarehouse, "default.db", tableName)
@@ -275,14 +264,13 @@ func readRows(t *testing.T, rb *paimon.ReadBuilder) []row
{
}
// openTestTable creates a catalog, opens the simple_log_table, and returns
-// the table along with a cleanup function. Skips the test if the warehouse
-// does not exist.
+// the table along with a cleanup function. Missing fixture data fails the
test.
func openTestTable(t *testing.T) *paimon.Table {
t.Helper()
- warehouse := testWarehouse()
+ warehouse := testWarehouse(t)
if _, err := os.Stat(warehouse); os.IsNotExist(err) {
- t.Skipf("Skipping: warehouse %s does not exist (run 'make
docker-up' first)", warehouse)
+ t.Fatalf("Missing test warehouse: %s", warehouse)
}
return openTableAt(t, warehouse, "simple_log_table")
}
@@ -441,11 +429,8 @@ func TestWriteOverwriteUsesBuilderMode(t *testing.T) {
}
func TestOverwriteRetrySameIdentifierIsIdempotent(t *testing.T) {
- warehouse := testWarehouse()
+ warehouse := testWarehouse(t)
source := filepath.Join(warehouse, "default.db", "simple_pk_table")
- if _, err := os.Stat(source); os.IsNotExist(err) {
- t.Skipf("Skipping: table %s does not exist (run 'make
docker-up' first)", source)
- }
targetWarehouse := t.TempDir()
target := filepath.Join(targetWarehouse, "default.db",
"simple_pk_table")
if err := copyDirectory(source, target); err != nil {
diff --git a/bindings/go/tests/resource_test.go
b/bindings/go/tests/resource_test.go
index 83c6e0cf..c4a76654 100644
--- a/bindings/go/tests/resource_test.go
+++ b/bindings/go/tests/resource_test.go
@@ -21,7 +21,6 @@ package paimon_test
import (
"errors"
- "os"
"path/filepath"
"strings"
"testing"
@@ -165,10 +164,6 @@ func TestWriteBuildersShareResourceBudget(t *testing.T) {
}
func TestPostponeWriteBuilderWithResourceBudget(t *testing.T) {
- warehouse := testWarehouse()
- if _, err := os.Stat(filepath.Join(warehouse, "default.db",
"postpone_fixed_bucket_pk_table")); os.IsNotExist(err) {
- t.Skip("postpone fixed-bucket test table is unavailable")
- }
table := openCopiedTable(t, "postpone_fixed_bucket_pk_table")
resources, err := paimon.NewResourceContext(0)
if err != nil {
diff --git a/bindings/go/tests/warehouse_test.go
b/bindings/go/tests/warehouse_test.go
new file mode 100644
index 00000000..4e0345e3
--- /dev/null
+++ b/bindings/go/tests/warehouse_test.go
@@ -0,0 +1,81 @@
+/*
+ * 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.
+ */
+
+package paimon_test
+
+import (
+ "archive/tar"
+ "compress/gzip"
+ "io"
+ "os"
+ "path/filepath"
+ "testing"
+)
+
+func testWarehouse(t *testing.T) string {
+ t.Helper()
+ if warehouse, ok := os.LookupEnv("PAIMON_TEST_WAREHOUSE"); ok {
+ return warehouse
+ }
+
+ // go test runs each package in its source directory, including with
-trimpath.
+ archivePath := filepath.Join("..", "..", "..", "crates",
"integrations", "datafusion", "testdata", "warehouse", "warehouse.tar.gz")
+ file, err := os.Open(archivePath)
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer file.Close()
+ decoder, err := gzip.NewReader(file)
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer decoder.Close()
+
+ warehouse := t.TempDir()
+ archive := tar.NewReader(decoder)
+ for {
+ entry, err := archive.Next()
+ if err == io.EOF {
+ break
+ }
+ if err != nil {
+ t.Fatal(err)
+ }
+ if entry.Typeflag != tar.TypeReg ||
!filepath.IsLocal(entry.Name) {
+ t.Fatalf("Unexpected golden archive entry: %s",
entry.Name)
+ }
+ target := filepath.Join(warehouse,
filepath.FromSlash(entry.Name))
+ if err := os.MkdirAll(filepath.Dir(target), 0755); err != nil {
+ t.Fatal(err)
+ }
+ output, err := os.Create(target)
+ if err != nil {
+ t.Fatal(err)
+ }
+ _, copyErr := io.Copy(output, archive)
+ closeErr := output.Close()
+ if copyErr != nil {
+ t.Fatal(copyErr)
+ }
+ if closeErr != nil {
+ t.Fatal(closeErr)
+ }
+ }
+ return warehouse
+}
diff --git a/bindings/python/README.md b/bindings/python/README.md
index 696e746a..fd549a08 100644
--- a/bindings/python/README.md
+++ b/bindings/python/README.md
@@ -528,11 +528,11 @@ make build
## Test
-Python integration tests expect the shared Paimon test warehouse to be prepared
-first from the repository root:
+Python integration tests automatically extract the committed golden warehouse
+into a temporary directory. Set `PAIMON_TEST_WAREHOUSE` to use an existing
+warehouse instead. Spark and Docker are not required.
```shell
-make docker-up
cd bindings/python
```
diff --git a/bindings/python/tests/test_datafusion.py
b/bindings/python/tests/test_datafusion.py
index 4fc8cfc8..6589c3d4 100644
--- a/bindings/python/tests/test_datafusion.py
+++ b/bindings/python/tests/test_datafusion.py
@@ -22,6 +22,7 @@ import os
import re
import struct
import sys
+import tarfile
import tempfile
import types
from pathlib import Path
@@ -38,7 +39,49 @@ from pypaimon_rust.datafusion import (
udf,
)
-WAREHOUSE = os.environ.get("PAIMON_TEST_WAREHOUSE", "/tmp/paimon-warehouse")
+
[email protected](scope="module")
+def golden_warehouse(tmp_path_factory):
+ override = os.environ.get("PAIMON_TEST_WAREHOUSE")
+ if override is not None:
+ return override
+
+ warehouse = tmp_path_factory.mktemp("golden-warehouse")
+ archive = (
+ Path(__file__).resolve().parents[3]
+ / "crates/integrations/datafusion/testdata/warehouse/warehouse.tar.gz"
+ )
+ with tarfile.open(archive, "r:gz") as fixture:
+ if hasattr(tarfile, "data_filter"):
+ fixture.extraction_filter = tarfile.data_filter
+ members = fixture.getmembers()
+ for member in members:
+ target = (warehouse / member.name).resolve()
+ if not member.isfile() or not target.is_relative_to(warehouse):
+ raise ValueError(f"Unexpected golden archive entry:
{member.name}")
+ fixture.extractall(warehouse, members=members)
+ return str(warehouse)
+
+
+def test_golden_warehouse_with_legacy_tarfile(monkeypatch, tmp_path_factory):
+ monkeypatch.delenv("PAIMON_TEST_WAREHOUSE", raising=False)
+ if hasattr(tarfile, "data_filter"):
+ original_extractall = tarfile.TarFile.extractall
+ trusted_filter = tarfile.fully_trusted_filter
+
+ def legacy_extractall(self, path=".", members=None, *,
numeric_owner=False):
+ return original_extractall(
+ self, path, members, numeric_owner=numeric_owner,
filter=trusted_filter
+ )
+
+ monkeypatch.setattr(tarfile.TarFile, "extractall", legacy_extractall)
+ monkeypatch.delattr(tarfile, "data_filter")
+
+ warehouse = golden_warehouse.__wrapped__(tmp_path_factory)
+ catalog = PaimonCatalog({"warehouse": warehouse})
+ assert
catalog.get_table("default.simple_log_table").latest_snapshot().id() == 1
+
+
PNG_SIGNATURE = b"\x89PNG\r\n\x1a\n"
BLOB_DESCRIPTOR_MAGIC = 0x424C4F4244455343
@@ -544,8 +587,8 @@ def test_vector_json_bridge_functions():
assert json.loads(row["vector_json"]) == [1.0, 2.5]
-def test_query_simple_table_via_catalog_provider():
- catalog = PaimonCatalog({"warehouse": WAREHOUSE})
+def test_query_simple_table_via_catalog_provider(golden_warehouse):
+ catalog = PaimonCatalog({"warehouse": golden_warehouse})
ctx = SessionContext()
ctx.register_catalog_provider("paimon", catalog)
@@ -980,8 +1023,8 @@ def test_register_catalog_discovers_pyjindo(monkeypatch,
tmp_path):
assert str(library) in str(error.value)
-def test_list_databases_and_tables():
- catalog = PaimonCatalog({"warehouse": WAREHOUSE})
+def test_list_databases_and_tables(golden_warehouse):
+ catalog = PaimonCatalog({"warehouse": golden_warehouse})
assert "default" in catalog.list_databases()
@@ -1000,8 +1043,8 @@ def test_list_databases_and_tables():
assert schema.partition_keys() == []
# ---------------- #285: observability ----------------
-def test_snapshots_for_simple_table():
- catalog = PaimonCatalog({"warehouse": WAREHOUSE})
+def test_snapshots_for_simple_table(golden_warehouse):
+ catalog = PaimonCatalog({"warehouse": golden_warehouse})
table = catalog.get_table("default.simple_log_table")
snap = table.latest_snapshot()
@@ -1016,8 +1059,8 @@ def test_snapshots_for_simple_table():
assert snaps[0].id() == snap.id()
-def test_partitions_and_tags_smoke():
- catalog = PaimonCatalog({"warehouse": WAREHOUSE})
+def test_partitions_and_tags_smoke(golden_warehouse):
+ catalog = PaimonCatalog({"warehouse": golden_warehouse})
table = catalog.get_table("default.simple_log_table")
# Non-partitioned, non-tagged table: both should be empty but well-typed.
diff --git a/crates/integration_tests/Cargo.toml
b/crates/integration_tests/Cargo.toml
index c93bfe79..75ec6617 100644
--- a/crates/integration_tests/Cargo.toml
+++ b/crates/integration_tests/Cargo.toml
@@ -33,5 +33,8 @@ tokio = { version = "1", features = ["macros",
"rt-multi-thread"] }
futures = "0.3"
[dev-dependencies]
+flate2 = "1"
+tar = "0.4"
+tempfile = "3"
serde_json = "1"
axum = { version = "0.7", features = ["macros", "tokio", "http1", "http2"] }
diff --git a/crates/integration_tests/tests/read_tables.rs
b/crates/integration_tests/tests/read_tables.rs
index 7de3ea71..2ad1c816 100644
--- a/crates/integration_tests/tests/read_tables.rs
+++ b/crates/integration_tests/tests/read_tables.rs
@@ -15,30 +15,28 @@
// specific language governing permissions and limitations
// under the License.
-//! Integration tests for reading Paimon tables provisioned by Spark.
+//! Integration tests for reading the committed Java-written golden warehouse.
use arrow_array::{
Array, ArrowPrimitiveType, Int32Array, Int64Array, ListArray, MapArray,
PrimitiveArray,
RecordBatch, StringArray, StructArray,
};
-use arrow_schema::{DataType as ArrowDataType, Field as ArrowField, Schema as
ArrowSchema};
use futures::TryStreamExt;
use paimon::api::ConfigResponse;
use paimon::catalog::{Identifier, RESTCatalog};
use paimon::common::Options;
use paimon::spec::{DataType, IntType, Predicate, Schema, VarCharType};
-use paimon::{Catalog, CatalogOptions, Error, FileSystemCatalog, Plan,
ScanStats};
+use paimon::{Catalog, Error, Plan, ScanStats};
use std::collections::{HashMap, HashSet};
-use std::sync::{Arc, OnceLock};
+
+#[path = "../../integrations/datafusion/testdata/warehouse/mod.rs"]
+mod golden_warehouse;
+use golden_warehouse::{test_catalog as create_file_system_catalog,
test_warehouse, TestWarehouse};
#[path = "../../paimon/tests/mock_server.rs"]
mod mock_server;
use mock_server::start_mock_server;
-fn get_test_warehouse() -> String {
- std::env::var("PAIMON_TEST_WAREHOUSE").unwrap_or_else(|_|
"/tmp/paimon-warehouse".to_string())
-}
-
async fn scan_and_read<C: Catalog + ?Sized>(
catalog: &C,
table_name: &str,
@@ -76,172 +74,17 @@ async fn get_table_from_catalog<C: Catalog + ?Sized>(
table_name: &str,
) -> paimon::Table {
let identifier = Identifier::new("default", table_name);
- match catalog.get_table(&identifier).await {
- Ok(table) => table,
- Err(Error::TableNotExist { .. }) if table_name ==
"data_evolution_table" => {
- ensure_data_evolution_table(catalog).await;
- catalog
- .get_table(&identifier)
- .await
- .expect("Failed to get generated data_evolution_table")
- }
- Err(err) => panic!("Failed to get table: {err:?}"),
- }
-}
-
-const DATA_EVOLUTION_FIXTURE_MARKER: &str = "paimon-rust.integration-fixture";
-const DATA_EVOLUTION_FIXTURE_VERSION: &str = "data-evolution-limit-v2";
-
-static DATA_EVOLUTION_FIXTURE_LOCK: OnceLock<tokio::sync::Mutex<()>> =
OnceLock::new();
-
-async fn ensure_data_evolution_table<C: Catalog + ?Sized>(catalog: &C) {
- let lock = DATA_EVOLUTION_FIXTURE_LOCK.get_or_init(||
tokio::sync::Mutex::new(()));
- let _guard = lock.lock().await;
-
- let identifier = Identifier::new("default", "data_evolution_table");
- if let Ok(table) = catalog.get_table(&identifier).await {
- let options = table.schema().options();
- let fixture_marker = options
- .get(DATA_EVOLUTION_FIXTURE_MARKER)
- .map(String::as_str);
- if fixture_marker == Some(DATA_EVOLUTION_FIXTURE_VERSION) ||
fixture_marker.is_none() {
- return;
- }
- catalog
- .drop_table(&identifier, true)
- .await
- .expect("Failed to replace stale generated data_evolution_table
fixture");
- }
-
- catalog
- .create_database("default", true, HashMap::new())
- .await
- .expect("Failed to create default database for data_evolution_table
fixture");
catalog
- .create_table(&identifier, data_evolution_fixture_schema(), true)
- .await
- .expect("Failed to create data_evolution_table fixture");
-
- let table = catalog
.get_table(&identifier)
.await
- .expect("Failed to load data_evolution_table fixture");
-
- append_data_evolution_batch(
- &table,
- data_evolution_fixture_batch(
- vec![1, 2, 3],
- vec!["alice", "bob", "carol"],
- vec![100, 200, 300],
- ),
- )
- .await;
- append_data_evolution_batch(
- &table,
- data_evolution_fixture_batch(vec![4, 5], vec!["dave", "eve"],
vec![400, 500]),
- )
- .await;
-
- let wb = table.new_write_builder();
- let mut update = wb
- .new_data_evolution_writer(vec!["name".to_string()])
- .expect("Failed to create data-evolution update writer");
- update
- .add_matched_batch(data_evolution_update_batch())
- .expect("Failed to add data-evolution update batch");
- wb.new_commit()
- .commit(
- update
- .prepare_commit()
- .await
- .expect("Failed to prepare data-evolution update commit"),
- )
- .await
- .expect("Failed to commit data-evolution update fixture");
-}
-
-fn data_evolution_fixture_schema() -> Schema {
- Schema::builder()
- .column("id", DataType::Int(IntType::new()))
- .column(
- "name",
-
DataType::VarChar(VarCharType::new(VarCharType::MAX_LENGTH).unwrap()),
- )
- .column("value", DataType::Int(IntType::new()))
- .option("row-tracking.enabled", "true")
- .option("data-evolution.enabled", "true")
- .option("source.split.target-size", "1b")
- .option("source.split.open-file-cost", "1b")
- .option(
- DATA_EVOLUTION_FIXTURE_MARKER,
- DATA_EVOLUTION_FIXTURE_VERSION,
- )
- .build()
- .unwrap()
-}
-
-fn data_evolution_fixture_batch(ids: Vec<i32>, names: Vec<&str>, values:
Vec<i32>) -> RecordBatch {
- let schema = Arc::new(ArrowSchema::new(vec![
- ArrowField::new("id", ArrowDataType::Int32, false),
- ArrowField::new("name", ArrowDataType::Utf8, false),
- ArrowField::new("value", ArrowDataType::Int32, false),
- ]));
- RecordBatch::try_new(
- schema,
- vec![
- Arc::new(Int32Array::from(ids)),
- Arc::new(StringArray::from(names)),
- Arc::new(Int32Array::from(values)),
- ],
- )
- .unwrap()
-}
-
-fn data_evolution_update_batch() -> RecordBatch {
- let schema = Arc::new(ArrowSchema::new(vec![
- ArrowField::new("_ROW_ID", ArrowDataType::Int64, false),
- ArrowField::new("name", ArrowDataType::Utf8, false),
- ]));
- RecordBatch::try_new(
- schema,
- vec![
- Arc::new(Int64Array::from(vec![0, 2, 3])),
- Arc::new(StringArray::from(vec!["alice-v2", "carol-v2", "dave"])),
- ],
- )
- .unwrap()
-}
-
-async fn append_data_evolution_batch(table: &paimon::Table, batch:
RecordBatch) {
- let wb = table.new_write_builder();
- let mut write = wb.new_write().expect("Failed to create fixture writer");
- write
- .write_arrow_batch(&batch)
- .await
- .expect("Failed to write fixture batch");
- wb.new_commit()
- .commit(
- write
- .prepare_commit()
- .await
- .expect("Failed to prepare fixture commit"),
- )
- .await
- .expect("Failed to commit fixture batch");
-}
-
-fn create_file_system_catalog() -> FileSystemCatalog {
- let warehouse = get_test_warehouse();
- let mut options = Options::new();
- options.set(CatalogOptions::WAREHOUSE, warehouse);
- FileSystemCatalog::new(options).expect("Failed to create
FileSystemCatalog")
+ .unwrap_or_else(|err| panic!("Failed to get golden table {table_name}:
{err:?}"))
}
async fn scan_and_read_with_fs_catalog(
table_name: &str,
projection: Option<&[&str]>,
) -> (Plan, Vec<RecordBatch>) {
- let catalog = create_file_system_catalog();
+ let (_warehouse, catalog) = create_file_system_catalog();
scan_and_read(&catalog, table_name, projection).await
}
@@ -636,7 +479,7 @@ async fn test_read_with_column_projection() {
#[tokio::test]
async fn test_read_projection_empty() {
- let catalog = create_file_system_catalog();
+ let (_warehouse, catalog) = create_file_system_catalog();
let table = get_table_from_catalog(&catalog, "simple_log_table").await;
let mut read_builder = table.new_read_builder();
@@ -680,7 +523,7 @@ async fn test_read_projection_empty() {
#[tokio::test]
async fn test_read_projection_unknown_column() {
- let catalog = create_file_system_catalog();
+ let (_warehouse, catalog) = create_file_system_catalog();
let table = get_table_from_catalog(&catalog, "simple_log_table").await;
let mut read_builder = table.new_read_builder();
@@ -702,7 +545,7 @@ async fn test_read_projection_unknown_column() {
#[tokio::test]
async fn test_read_projection_all_invalid() {
- let catalog = create_file_system_catalog();
+ let (_warehouse, catalog) = create_file_system_catalog();
let table = get_table_from_catalog(&catalog, "simple_log_table").await;
let mut read_builder = table.new_read_builder();
@@ -724,7 +567,7 @@ async fn test_read_projection_all_invalid() {
#[tokio::test]
async fn test_read_projection_duplicate_column() {
- let catalog = create_file_system_catalog();
+ let (_warehouse, catalog) = create_file_system_catalog();
let table = get_table_from_catalog(&catalog, "simple_log_table").await;
// `id` exists, so with_projection passes its best-effort early check; the
@@ -752,7 +595,7 @@ async fn test_read_projection_duplicate_column() {
async fn test_read_partitioned_table_with_filter() {
use paimon::spec::{Datum, PredicateBuilder};
- let catalog = create_file_system_catalog();
+ let (_warehouse, catalog) = create_file_system_catalog();
let table = get_table_from_catalog(&catalog,
"partitioned_log_table").await;
let schema = table.schema();
let pb = PredicateBuilder::new(schema.fields());
@@ -782,7 +625,7 @@ async fn test_read_partitioned_table_with_filter() {
async fn test_read_multi_partitioned_table_with_filter() {
use paimon::spec::{Datum, Predicate, PredicateBuilder};
- let catalog = create_file_system_catalog();
+ let (_warehouse, catalog) = create_file_system_catalog();
let table = get_table_from_catalog(&catalog,
"multi_partitioned_log_table").await;
let schema = table.schema();
let pb = PredicateBuilder::new(schema.fields());
@@ -812,7 +655,7 @@ async fn test_read_multi_partitioned_table_with_filter() {
async fn test_read_partitioned_table_data_only_filter_prunes_all_files() {
use paimon::spec::{Datum, PredicateBuilder};
- let catalog = create_file_system_catalog();
+ let (_warehouse, catalog) = create_file_system_catalog();
let table = get_table_from_catalog(&catalog,
"partitioned_log_table").await;
let schema = table.schema();
let pb = PredicateBuilder::new(schema.fields());
@@ -841,7 +684,7 @@ async fn
test_read_partitioned_table_data_only_filter_prunes_all_files() {
async fn test_read_partitioned_table_mixed_and_filter() {
use paimon::spec::{Datum, Predicate, PredicateBuilder};
- let catalog = create_file_system_catalog();
+ let (_warehouse, catalog) = create_file_system_catalog();
let table = get_table_from_catalog(&catalog,
"partitioned_log_table").await;
let schema = table.schema();
let pb = PredicateBuilder::new(schema.fields());
@@ -871,7 +714,7 @@ async fn test_read_partitioned_table_mixed_and_filter() {
async fn
test_read_partitioned_table_data_only_filter_keeps_matching_partition() {
use paimon::spec::{Datum, PredicateBuilder};
- let catalog = create_file_system_catalog();
+ let (_warehouse, catalog) = create_file_system_catalog();
let table = get_table_from_catalog(&catalog,
"partitioned_log_table").await;
let schema = table.schema();
let pb = PredicateBuilder::new(schema.fields());
@@ -902,7 +745,7 @@ async fn
test_read_partitioned_table_data_only_filter_keeps_matching_partition()
async fn
test_read_multi_partitioned_table_or_of_mixed_ands_prunes_partitions() {
use paimon::spec::{Datum, Predicate, PredicateBuilder};
- let catalog = create_file_system_catalog();
+ let (_warehouse, catalog) = create_file_system_catalog();
let table = get_table_from_catalog(&catalog,
"multi_partitioned_log_table").await;
let schema = table.schema();
let pb = PredicateBuilder::new(schema.fields());
@@ -946,7 +789,7 @@ async fn
test_read_multi_partitioned_table_or_of_mixed_ands_prunes_partitions()
async fn test_read_partitioned_table_mixed_or_filter_preserves_all() {
use paimon::spec::{Datum, Predicate, PredicateBuilder};
- let catalog = create_file_system_catalog();
+ let (_warehouse, catalog) = create_file_system_catalog();
let table = get_table_from_catalog(&catalog,
"partitioned_log_table").await;
let schema = table.schema();
let pb = PredicateBuilder::new(schema.fields());
@@ -981,7 +824,7 @@ async fn
test_read_partitioned_table_mixed_or_filter_preserves_all() {
async fn test_read_partitioned_table_filter_matches_no_partition() {
use paimon::spec::{Datum, PredicateBuilder};
- let catalog = create_file_system_catalog();
+ let (_warehouse, catalog) = create_file_system_catalog();
let table = get_table_from_catalog(&catalog,
"partitioned_log_table").await;
let schema = table.schema();
let pb = PredicateBuilder::new(schema.fields());
@@ -1007,7 +850,7 @@ async fn
test_read_partitioned_table_filter_matches_no_partition() {
async fn test_read_partitioned_table_eval_row_error_fails_open() {
use paimon::spec::{ArrayType, DataType, Datum, IntType, PredicateOperator};
- let catalog = create_file_system_catalog();
+ let (_warehouse, catalog) = create_file_system_catalog();
let table = get_table_from_catalog(&catalog,
"partitioned_log_table").await;
let dt_index = table
.schema()
@@ -1046,7 +889,7 @@ async fn
test_read_partitioned_table_eval_row_error_fails_open() {
// ======================= REST Catalog read tests
===============================
-/// Build a simple test schema matching the Spark-provisioned tables (id INT,
name VARCHAR).
+/// Build a simple test schema matching the golden tables (id INT, name
VARCHAR).
fn simple_log_schema() -> Schema {
Schema::builder()
.column("id", DataType::Int(IntType::new()))
@@ -1055,12 +898,13 @@ fn simple_log_schema() -> Schema {
.expect("Failed to build schema")
}
-/// Start a mock REST server backed by Spark-provisioned data on disk,
+/// Start a mock REST server backed by golden data on disk,
/// register the given tables, and return a connected `RESTCatalog`.
async fn setup_rest_catalog_with_tables(
table_configs: &[(&str, &str, Schema)],
-) -> (mock_server::RESTServer, RESTCatalog) {
- let catalog_path = get_test_warehouse();
+) -> (TestWarehouse, mock_server::RESTServer, RESTCatalog) {
+ let warehouse = test_warehouse();
+ let catalog_path = warehouse.path().to_string();
// Use a simple warehouse name (no slashes) to avoid URL-encoding issues
let warehouse_name = "test_warehouse";
let prefix = "mock-test";
@@ -1093,16 +937,16 @@ async fn setup_rest_catalog_with_tables(
.await
.expect("Failed to create RESTCatalog");
- (server, catalog)
+ (warehouse, server, catalog)
}
/// Test reading an append-only (log) table via REST catalog backed by mock
server.
///
-/// The mock server returns table metadata pointing to Spark-provisioned data
on disk.
+/// The mock server returns table metadata pointing to golden data on disk.
#[tokio::test]
async fn test_rest_catalog_read_append_table() {
let table_name = "simple_log_table";
- let (_server, catalog) =
+ let (_warehouse, _server, catalog) =
setup_rest_catalog_with_tables(&[("default", table_name,
simple_log_schema())]).await;
let (plan, batches) = scan_and_read(&catalog, table_name, None).await;
@@ -1135,7 +979,7 @@ async fn test_rest_catalog_read_append_table() {
/// Test reading a data-evolution enabled append-only table.
///
-/// The table is provisioned by Spark with `data-evolution.enabled=true` and
+/// The table is stored in the golden warehouse with
`data-evolution.enabled=true` and
/// `row-tracking.enabled=true`. Multiple inserts produce files with
`first_row_id`
/// set, exercising the data evolution scan and read path.
#[tokio::test]
@@ -1254,7 +1098,7 @@ async fn plan_table_with_stats(table: &paimon::Table,
limit: Option<usize>) -> (
/// Test limit pushdown: when limit is smaller than total rows, fewer data
files may be generated.
#[tokio::test]
async fn test_limit_pushdown() {
- let catalog = create_file_system_catalog();
+ let (_warehouse, catalog) = create_file_system_catalog();
// Test limit pushdown for data evolution table
let table = get_table_from_catalog(&catalog, "data_evolution_table").await;
@@ -1424,7 +1268,7 @@ async fn test_read_format_schema_evolution_add_column() {
use paimon::spec::{Datum, PredicateBuilder};
let table_name = "format_schema_evolution_add_column";
- let catalog = create_file_system_catalog();
+ let (_warehouse, catalog) = create_file_system_catalog();
let table = get_table_from_catalog(&catalog, table_name).await;
let (plan, batches) = scan_and_read(&catalog, table_name, None).await;
assert_plan_file_formats(&plan, &["avro", "orc", "parquet"], table_name);
@@ -1494,7 +1338,7 @@ async fn
test_read_partitioned_format_schema_evolution_add_column() {
use paimon::spec::{Datum, Predicate, PredicateBuilder};
let table_name = "partitioned_format_schema_evolution_add_column";
- let catalog = create_file_system_catalog();
+ let (_warehouse, catalog) = create_file_system_catalog();
let table = get_table_from_catalog(&catalog, table_name).await;
let (plan, batches) = scan_and_read(&catalog, table_name, None).await;
assert_plan_file_formats(&plan, &["avro", "orc", "parquet"], table_name);
@@ -1626,7 +1470,7 @@ async fn
test_read_format_schema_evolution_type_promotion() {
use paimon::spec::{Datum, PredicateBuilder};
let table_name = "format_schema_evolution_type_promotion";
- let catalog = create_file_system_catalog();
+ let (_warehouse, catalog) = create_file_system_catalog();
let table = get_table_from_catalog(&catalog, table_name).await;
let (plan, batches) = scan_and_read(&catalog, table_name, None).await;
assert_plan_file_formats(&plan, &["avro", "orc", "parquet"], table_name);
@@ -1688,7 +1532,7 @@ async fn
test_read_format_schema_evolution_type_promotion() {
async fn
test_stats_pruning_schema_evolution_added_column_eq_prunes_old_files() {
use paimon::spec::{Datum, PredicateBuilder};
- let catalog = create_file_system_catalog();
+ let (_warehouse, catalog) = create_file_system_catalog();
let table = get_table_from_catalog(&catalog,
"schema_evolution_add_column").await;
let pb = PredicateBuilder::new(table.schema().fields());
let filter = pb
@@ -1715,7 +1559,7 @@ async fn
test_stats_pruning_schema_evolution_added_column_eq_prunes_old_files()
async fn
test_stats_pruning_schema_evolution_added_column_is_null_prunes_new_files() {
use paimon::spec::PredicateBuilder;
- let catalog = create_file_system_catalog();
+ let (_warehouse, catalog) = create_file_system_catalog();
let table = get_table_from_catalog(&catalog,
"schema_evolution_add_column").await;
let pb = PredicateBuilder::new(table.schema().fields());
let filter = pb.is_null("age").expect("Failed to build predicate");
@@ -1740,7 +1584,7 @@ async fn
test_stats_pruning_schema_evolution_added_column_is_null_prunes_new_fil
async fn
test_stats_pruning_schema_evolution_type_promotion_prunes_old_int_files() {
use paimon::spec::{Datum, PredicateBuilder};
- let catalog = create_file_system_catalog();
+ let (_warehouse, catalog) = create_file_system_catalog();
let table = get_table_from_catalog(&catalog,
"schema_evolution_type_promotion").await;
let pb = PredicateBuilder::new(table.schema().fields());
let filter = pb
@@ -2008,7 +1852,7 @@ async fn test_read_schema_evolution_drop_column() {
async fn test_read_schema_evolution_rename_column() {
use paimon::spec::{Datum, PredicateBuilder};
- let catalog = create_file_system_catalog();
+ let (_warehouse, catalog) = create_file_system_catalog();
let table_name = "schema_evolution_rename_column";
let table = get_table_from_catalog(&catalog, table_name).await;
let (plan, batches) = scan_and_read(&catalog, table_name, None).await;
@@ -2096,7 +1940,7 @@ async fn
test_read_mixed_format_schema_evolution_drop_column() {
use paimon::spec::{Datum, PredicateBuilder};
let table_name = "mixed_format_schema_evolution_drop_column";
- let catalog = create_file_system_catalog();
+ let (_warehouse, catalog) = create_file_system_catalog();
let table = get_table_from_catalog(&catalog, table_name).await;
let (plan, batches) = scan_and_read(&catalog, table_name, None).await;
assert_plan_file_formats(&plan, &["avro", "orc", "parquet"], table_name);
@@ -2200,7 +2044,7 @@ async fn
test_read_mixed_format_schema_evolution_reorder_move_column() {
use paimon::spec::{Datum, PredicateBuilder};
let table_name = "mixed_format_schema_evolution_reorder_move_column";
- let catalog = create_file_system_catalog();
+ let (_warehouse, catalog) = create_file_system_catalog();
let table = get_table_from_catalog(&catalog, table_name).await;
let (plan, batches) = scan_and_read(&catalog, table_name, None).await;
@@ -2431,7 +2275,7 @@ async fn test_read_complex_type_table() {
/// Reading a primary-key table without deletion vectors should work via
sort-merge reader.
#[tokio::test]
async fn test_read_pk_table_without_dv_via_sort_merge() {
- let catalog = create_file_system_catalog();
+ let (_warehouse, catalog) = create_file_system_catalog();
let table = get_table_from_catalog(&catalog, "simple_pk_table").await;
let read_builder = table.new_read_builder();
@@ -2472,7 +2316,7 @@ async fn test_read_pk_table_without_dv_via_sort_merge() {
/// The table has been compacted so all files are level > 0, and the scan
skips level-0 files.
#[tokio::test]
async fn test_read_first_row_pk_table() {
- let catalog = create_file_system_catalog();
+ let (_warehouse, catalog) = create_file_system_catalog();
let table = get_table_from_catalog(&catalog, "first_row_pk_table").await;
let read_builder = table.new_read_builder();
@@ -2534,7 +2378,7 @@ async fn test_read_non_pk_table_with_dv() {
/// and should NOT be visible to batch readers. The plan should produce no
splits.
#[tokio::test]
async fn test_read_postpone_bucket_pk_table_returns_empty() {
- let catalog = create_file_system_catalog();
+ let (_warehouse, catalog) = create_file_system_catalog();
let table = get_table_from_catalog(&catalog,
"postpone_bucket_pk_table").await;
let read_builder = table.new_read_builder();
@@ -2557,7 +2401,7 @@ async fn
test_read_postpone_bucket_pk_table_returns_empty() {
async fn test_data_evolution_table_with_filter() {
use paimon::spec::{Datum, PredicateBuilder};
- let catalog = create_file_system_catalog();
+ let (_warehouse, catalog) = create_file_system_catalog();
let table = get_table_from_catalog(&catalog, "data_evolution_table").await;
let pb = PredicateBuilder::new(table.schema().fields());
@@ -2616,7 +2460,7 @@ async fn test_data_evolution_table_with_filter() {
async fn test_bucket_predicate_filtering() {
use paimon::spec::{Datum, PredicateBuilder};
- let catalog = create_file_system_catalog();
+ let (_warehouse, catalog) = create_file_system_catalog();
let table = get_table_from_catalog(&catalog,
"multi_bucket_pk_table").await;
let schema = table.schema();
let pb = PredicateBuilder::new(schema.fields());
@@ -2668,7 +2512,7 @@ async fn test_bucket_predicate_filtering() {
async fn test_bucket_predicate_filtering_in() {
use paimon::spec::{Datum, PredicateBuilder};
- let catalog = create_file_system_catalog();
+ let (_warehouse, catalog) = create_file_system_catalog();
let table = get_table_from_catalog(&catalog,
"multi_bucket_pk_table").await;
let schema = table.schema();
let pb = PredicateBuilder::new(schema.fields());
@@ -2701,7 +2545,7 @@ async fn test_bucket_predicate_filtering_in() {
/// Time travel by snapshot id: snapshot 1 should return only the first batch.
#[tokio::test]
async fn test_time_travel_by_snapshot_id() {
- let catalog = create_file_system_catalog();
+ let (_warehouse, catalog) = create_file_system_catalog();
let table = get_table_from_catalog(&catalog, "time_travel_table").await;
// Snapshot 1: (1, 'alice'), (2, 'bob')
@@ -2755,7 +2599,7 @@ async fn test_time_travel_by_snapshot_id() {
/// Time travel by tag name.
#[tokio::test]
async fn test_time_travel_by_tag_name() {
- let catalog = create_file_system_catalog();
+ let (_warehouse, catalog) = create_file_system_catalog();
let table = get_table_from_catalog(&catalog, "time_travel_table").await;
// Tag 'snapshot1' -> snapshot 1: (1, 'alice'), (2, 'bob')
@@ -2808,7 +2652,7 @@ async fn test_time_travel_by_tag_name() {
#[tokio::test]
async fn time_travel_schema_evolution() {
- let catalog = create_file_system_catalog();
+ let (_warehouse, catalog) = create_file_system_catalog();
let table = get_table_from_catalog(&catalog,
"time_travel_schema_evolution").await;
for (tag, expected_fields, expected_rows) in [
@@ -2957,7 +2801,7 @@ async fn time_travel_schema_evolution() {
#[tokio::test]
async fn test_time_travel_conflicting_selectors_fail() {
- let catalog = create_file_system_catalog();
+ let (_warehouse, catalog) = create_file_system_catalog();
let table = get_table_from_catalog(&catalog, "time_travel_table").await;
// Java resolves scan.version before validating conflicts with other
selectors.
@@ -2998,7 +2842,7 @@ async fn test_time_travel_conflicting_selectors_fail() {
#[tokio::test]
async fn test_time_travel_invalid_version_fails() {
- let catalog = create_file_system_catalog();
+ let (_warehouse, catalog) = create_file_system_catalog();
let table = get_table_from_catalog(&catalog, "time_travel_table").await;
let invalid = table.copy_with_options(HashMap::from([(
@@ -3090,7 +2934,7 @@ async fn test_read_data_evolution_drop_column() {
async fn test_limit_pushdown_disabled_with_data_predicates() {
use paimon::spec::{Datum, PredicateBuilder};
- let catalog = create_file_system_catalog();
+ let (_warehouse, catalog) = create_file_system_catalog();
let table = get_table_from_catalog(&catalog, "data_evolution_table").await;
let pb = PredicateBuilder::new(table.schema().fields());
@@ -3147,7 +2991,7 @@ fn extract_code_value(batches: &[RecordBatch]) ->
Vec<(String, i32)> {
async fn test_bucket_predicate_filtering_short_string_key() {
use paimon::spec::{Datum, PredicateBuilder};
- let catalog = create_file_system_catalog();
+ let (_warehouse, catalog) = create_file_system_catalog();
let table = get_table_from_catalog(&catalog,
"string_bucket_short_key").await;
let pb = PredicateBuilder::new(table.schema().fields());
@@ -3189,7 +3033,7 @@ async fn
test_bucket_predicate_filtering_short_string_key() {
async fn test_bucket_predicate_filtering_long_string_key() {
use paimon::spec::{Datum, PredicateBuilder};
- let catalog = create_file_system_catalog();
+ let (_warehouse, catalog) = create_file_system_catalog();
let table = get_table_from_catalog(&catalog,
"string_bucket_long_key").await;
let pb = PredicateBuilder::new(table.schema().fields());
@@ -3278,7 +3122,7 @@ fn extract_id_name_value(batches: &[RecordBatch]) ->
Vec<(i32, String, i32)> {
async fn test_read_data_evolution_table_with_row_ranges() {
use paimon::RowRange;
- let catalog = create_file_system_catalog();
+ let (_warehouse, catalog) = create_file_system_catalog();
let table = get_table_from_catalog(&catalog, "data_evolution_table").await;
let (full_plan, full_batches) = scan_and_read(&catalog,
"data_evolution_table", None).await;
@@ -3322,7 +3166,7 @@ async fn test_read_data_evolution_table_with_row_ranges()
{
async fn test_read_data_evolution_table_with_empty_row_ranges() {
use paimon::RowRange;
- let catalog = create_file_system_catalog();
+ let (_warehouse, catalog) = create_file_system_catalog();
let table = get_table_from_catalog(&catalog, "data_evolution_table").await;
let (plan, batches) =
@@ -3337,7 +3181,7 @@ async fn
test_read_data_evolution_table_with_empty_row_ranges() {
async fn test_read_data_evolution_table_with_full_row_ranges() {
use paimon::RowRange;
- let catalog = create_file_system_catalog();
+ let (_warehouse, catalog) = create_file_system_catalog();
let table = get_table_from_catalog(&catalog, "data_evolution_table").await;
let (_, full_batches) = scan_and_read(&catalog, "data_evolution_table",
None).await;
@@ -3354,7 +3198,7 @@ async fn
test_read_data_evolution_table_with_full_row_ranges() {
async fn test_limit_pushdown_disabled_with_row_ranges() {
use paimon::RowRange;
- let catalog = create_file_system_catalog();
+ let (_warehouse, catalog) = create_file_system_catalog();
let table = get_table_from_catalog(&catalog, "data_evolution_table").await;
let mut row_range_reader = table.new_read_builder();
@@ -3393,7 +3237,7 @@ async fn test_limit_pushdown_disabled_with_row_ranges() {
#[tokio::test]
async fn test_read_data_evolution_table_with_row_id_projection() {
- let catalog = create_file_system_catalog();
+ let (_warehouse, catalog) = create_file_system_catalog();
let table = get_table_from_catalog(&catalog, "data_evolution_table").await;
// Project _ROW_ID along with regular columns
@@ -3449,7 +3293,7 @@ async fn
test_read_data_evolution_table_with_row_id_projection() {
async fn test_read_data_evolution_table_only_row_id_with_row_ranges() {
use paimon::RowRange;
- let catalog = create_file_system_catalog();
+ let (_warehouse, catalog) = create_file_system_catalog();
let table = get_table_from_catalog(&catalog, "data_evolution_table").await;
// Get full row ID range
@@ -3501,7 +3345,7 @@ async fn
test_read_data_evolution_table_only_row_id_with_row_ranges() {
#[tokio::test]
async fn test_read_data_evolution_mixed_format_row_id_projection() {
- let catalog = create_file_system_catalog();
+ let (_warehouse, catalog) = create_file_system_catalog();
let table = get_table_from_catalog(&catalog,
"data_evolution_mixed_format_add_column").await;
let mut read_builder = table.new_read_builder();
@@ -3851,7 +3695,7 @@ async fn test_read_full_types_table() {
async fn test_read_orc_with_filter_only_column_projection() {
use paimon::spec::{Datum, PredicateBuilder};
- let catalog = create_file_system_catalog();
+ let (_warehouse, catalog) = create_file_system_catalog();
let table = get_table_from_catalog(&catalog, "full_types_table").await;
let pb = PredicateBuilder::new(table.schema().fields());
let filter = pb
@@ -3880,7 +3724,7 @@ async fn assert_full_types_orc_filter_matches(
projected_column: &str,
expected_string_values: &[&str],
) {
- let catalog = create_file_system_catalog();
+ let (_warehouse, catalog) = create_file_system_catalog();
let table = get_table_from_catalog(&catalog, "full_types_table").await;
let (_, batches) =
@@ -3904,7 +3748,7 @@ async fn assert_full_types_orc_filter_matches(
async fn test_read_orc_with_supported_predicate_pushdown_types() {
use paimon::spec::{Datum, PredicateBuilder};
- let catalog = create_file_system_catalog();
+ let (_warehouse, catalog) = create_file_system_catalog();
let table = get_table_from_catalog(&catalog, "full_types_table").await;
let pb = PredicateBuilder::new(table.schema().fields());
@@ -3976,7 +3820,7 @@ async fn
test_read_orc_with_supported_predicate_pushdown_types() {
async fn test_read_orc_with_unsupported_date_predicate_remains_residual() {
use paimon::spec::{Datum, PredicateBuilder};
- let catalog = create_file_system_catalog();
+ let (_warehouse, catalog) = create_file_system_catalog();
let table = get_table_from_catalog(&catalog, "full_types_table").await;
let pb = PredicateBuilder::new(table.schema().fields());
let filter = pb
@@ -3990,7 +3834,7 @@ async fn
test_read_orc_with_unsupported_date_predicate_remains_residual() {
async fn test_read_orc_predicate_pushdown_remains_conservative() {
use paimon::spec::{Datum, PredicateBuilder};
- let catalog = create_file_system_catalog();
+ let (_warehouse, catalog) = create_file_system_catalog();
let table = get_table_from_catalog(&catalog, "full_types_table").await;
let pb = PredicateBuilder::new(table.schema().fields());
let filter = pb.equal("id", Datum::Int(2)).expect("build id predicate");
diff --git a/crates/integrations/datafusion/src/lib.rs
b/crates/integrations/datafusion/src/lib.rs
index 84bbdd16..89910c17 100644
--- a/crates/integrations/datafusion/src/lib.rs
+++ b/crates/integrations/datafusion/src/lib.rs
@@ -49,6 +49,9 @@ mod format_partition_repair;
mod format_table_truncate;
#[cfg(feature = "fulltext")]
mod full_text_search;
+#[cfg(test)]
+#[path = "../testdata/warehouse/mod.rs"]
+mod golden_warehouse;
mod hybrid_search;
mod lateral_vector_search;
mod memory;
diff --git a/crates/integrations/datafusion/src/table/mod.rs
b/crates/integrations/datafusion/src/table/mod.rs
index a3ba2325..730cd30b 100644
--- a/crates/integrations/datafusion/src/table/mod.rs
+++ b/crates/integrations/datafusion/src/table/mod.rs
@@ -565,8 +565,9 @@ mod tests {
use datafusion::prelude::{SessionConfig, SessionContext};
use paimon::catalog::Identifier;
use paimon::spec::{ArrayType, MapType, RowType, VarCharType};
- use paimon::{Catalog, CatalogOptions, DataSplit, FileSystemCatalog,
Options};
+ use paimon::{Catalog, DataSplit};
+ use crate::golden_warehouse::{test_catalog as create_catalog,
TestWarehouse};
use crate::physical_plan::PaimonTableScan;
#[test]
@@ -587,27 +588,17 @@ mod tests {
assert_eq!(result, vec![vec![1, 2, 3]]);
}
- fn get_test_warehouse() -> String {
- std::env::var("PAIMON_TEST_WAREHOUSE")
- .unwrap_or_else(|_| "/tmp/paimon-warehouse".to_string())
- }
-
- fn create_catalog() -> FileSystemCatalog {
- let warehouse = get_test_warehouse();
- let mut options = Options::new();
- options.set(CatalogOptions::WAREHOUSE, warehouse);
- FileSystemCatalog::new(options).expect("Failed to create catalog")
- }
-
- async fn create_provider(table_name: &str) -> PaimonTableProvider {
- let catalog = create_catalog();
+ async fn create_provider(table_name: &str) -> (TestWarehouse,
PaimonTableProvider) {
+ let (warehouse, catalog) = create_catalog();
let identifier = Identifier::new("default", table_name);
let table = catalog
.get_table(&identifier)
.await
.expect("Failed to get table");
- PaimonTableProvider::try_new(table).expect("Failed to create table
provider")
+ let provider =
+ PaimonTableProvider::try_new(table).expect("Failed to create table
provider");
+ (warehouse, provider)
}
async fn plan_partitions(
@@ -937,7 +928,7 @@ mod tests {
#[tokio::test]
async fn test_scan_partition_filter_plans_matching_partition_set() {
- let provider = create_provider("partitioned_log_table").await;
+ let (_warehouse, provider) =
create_provider("partitioned_log_table").await;
let planned_partitions =
plan_partitions(&provider, vec![col("dt").eq(lit("2024-01-01"))],
None).await;
@@ -949,7 +940,7 @@ mod tests {
#[tokio::test]
async fn test_scan_mixed_and_filter_keeps_partition_pruning() {
- let provider = create_provider("partitioned_log_table").await;
+ let (_warehouse, provider) =
create_provider("partitioned_log_table").await;
let planned_partitions = plan_partitions(
&provider,
vec![col("dt").eq(lit("2024-01-01")).and(col("id").gt(lit(1)))],
@@ -965,7 +956,7 @@ mod tests {
#[tokio::test]
async fn test_scan_multi_partition_filter_plans_exact_partition_set() {
- let provider = create_provider("multi_partitioned_log_table").await;
+ let (_warehouse, provider) =
create_provider("multi_partitioned_log_table").await;
let dt_only_partitions =
plan_partitions(&provider, vec![col("dt").eq(lit("2024-01-01"))],
None).await;
@@ -991,7 +982,7 @@ mod tests {
#[tokio::test]
async fn
test_scan_partially_translated_not_filter_prunes_partitions_but_skips_limit_hint()
{
- let provider = create_provider("multi_partitioned_log_table").await;
+ let (_warehouse, provider) =
create_provider("multi_partitioned_log_table").await;
let filter = col("dt")
.eq(lit("2024-01-01"))
.and(Expr::Not(Box::new(col("hr").eq(lit(10)))));
@@ -1020,7 +1011,7 @@ mod tests {
#[tokio::test]
async fn test_scan_keeps_pushed_predicate_for_execute() {
- let provider = create_provider("partitioned_log_table").await;
+ let (_warehouse, provider) =
create_provider("partitioned_log_table").await;
let filter = col("id").gt(lit(1));
let config = SessionConfig::new().with_target_partitions(8);
@@ -1093,7 +1084,7 @@ mod tests {
#[tokio::test]
async fn test_scan_applies_limit_hint_only_when_safe() {
- let provider = create_provider("partitioned_log_table").await;
+ let (_warehouse, provider) =
create_provider("partitioned_log_table").await;
let full_plan = plan_partitions(&provider, vec![], None).await;
let plan = plan_scan(&provider, vec![], Some(1)).await;
let scan = plan
@@ -1115,7 +1106,7 @@ mod tests {
#[tokio::test]
async fn test_scan_keeps_limit_but_skips_limit_pruning_for_data_filters() {
- let provider = create_provider("partitioned_log_table").await;
+ let (_warehouse, provider) =
create_provider("partitioned_log_table").await;
let filter = col("id").gt(lit(1));
let full_plan = plan_partitions(&provider, vec![filter.clone()],
None).await;
let plan = plan_scan(&provider, vec![filter], Some(1)).await;
diff --git a/crates/integrations/datafusion/testdata/warehouse/README.md
b/crates/integrations/datafusion/testdata/warehouse/README.md
new file mode 100644
index 00000000..aaff2995
--- /dev/null
+++ b/crates/integrations/datafusion/testdata/warehouse/README.md
@@ -0,0 +1,64 @@
+<!--
+ ~ 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.
+-->
+
+# Java-written golden warehouse
+
+`warehouse.tar.gz` contains 34 Paimon tables under `default.db/`, shared by the
+Rust, DataFusion, Python, and Go integration tests. Each suite extracts its own
+temporary copy unless `PAIMON_TEST_WAREHOUSE` points to an existing warehouse.
+The tests need no Spark, Java, Docker, network download, or provisioning step.
+Rust readers retain an owned guard until their reads finish; concurrent readers
+share a weakly cached extraction that is deleted after the last guard drops.
+Write tests use private copies and must not modify this archive.
+
+## Provenance
+
+- Producer: Apache Spark 3.5.3 with Apache Paimon 1.3.1, `local[2]`, UTC.
+- Source: the former
+
[`dev/spark/provision.py`](https://github.com/apache/paimon-rust/blob/fb5e8f7efeb592fab916b8ee8873405a947e2f17/dev/spark/provision.py).
+- Generator SHA-256:
`54c4f866308bf5d2b1b9744315555d5305a3bb17ebf58c18fb4dee220265b8f7`.
+- Archive SHA-256:
`dd9bc2d03975955800ea5bc0b2b9ed39612c3461fba00708b3ff76f450cdac13`.
+- Contents: 658 files, 588,603 uncompressed bytes; 118,891 compressed bytes.
+
+The tables were produced through the Java writer's normal catalog, commit,
+compaction, and schema-evolution paths. File contents have not been rewritten.
+Only archive paths, permissions, ownership, entry order, and archive timestamps
+were normalized. All schemas, snapshots, tags, manifests, deletion vectors,
+bucket indexes, and data files are retained, including historical files needed
+for time travel. UUIDs and commit timestamps inside those files remain frozen.
+
+## Coverage
+
+- Append and primary-key tables, including deletion vectors, multiple buckets,
+ partitioned tables, and the first-row merge engine.
+- Dynamic, postponed, and postponed fixed buckets; short and long string keys.
+- Schema addition, type promotion, dropping, renaming, and column reordering.
+- Data evolution with row tracking, partial column updates, and mixed formats.
+- Snapshot and tag time travel across schema changes.
+- Parquet, ORC, and Avro; complex types, full type coverage, and boundary
values.
+
+Expected rows and query behavior remain explicit in the tests. A missing or
+invalid fixture must fail rather than falling back to data generated by the
+Rust implementation under test.
+
+When adding a cross-language compatibility case, generate the table with an
+independent Java Paimon writer, record its version and source, and add explicit
+behavioral assertions. Preserve existing fixtures when extending coverage for a
+new format version. Changes to UUIDs, timestamps, file counts, and split layout
+need review; matching row values alone does not preserve metadata coverage.
diff --git a/crates/integrations/datafusion/testdata/warehouse/mod.rs
b/crates/integrations/datafusion/testdata/warehouse/mod.rs
new file mode 100644
index 00000000..f4162b9a
--- /dev/null
+++ b/crates/integrations/datafusion/testdata/warehouse/mod.rs
@@ -0,0 +1,125 @@
+// 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.
+
+//! Scoped access to the shared Java-written warehouse.
+
+use std::sync::{Arc, Mutex, Weak};
+
+use paimon::{CatalogOptions, FileSystemCatalog, Options};
+
+pub enum TestWarehouse {
+ External(String),
+ Golden(Arc<tempfile::TempDir>),
+}
+
+impl TestWarehouse {
+ pub fn path(&self) -> &str {
+ match self {
+ Self::External(path) => path,
+ Self::Golden(directory) => directory.path().to_str().expect("UTF-8
warehouse path"),
+ }
+ }
+}
+
+pub fn test_catalog() -> (TestWarehouse, FileSystemCatalog) {
+ let warehouse = test_warehouse();
+ let mut options = Options::new();
+ options.set(CatalogOptions::WAREHOUSE, warehouse.path());
+ let catalog = FileSystemCatalog::new(options).expect("Failed to create
FileSystemCatalog");
+ (warehouse, catalog)
+}
+
+pub fn test_warehouse() -> TestWarehouse {
+ if let Ok(path) = std::env::var("PAIMON_TEST_WAREHOUSE") {
+ return TestWarehouse::External(path);
+ }
+
+ static WAREHOUSE: Mutex<Weak<tempfile::TempDir>> = Mutex::new(Weak::new());
+ unpack_warehouse(&WAREHOUSE)
+}
+
+fn unpack_warehouse(cache: &Mutex<Weak<tempfile::TempDir>>) -> TestWarehouse {
+ let mut cached = cache.lock().expect("Golden warehouse cache poisoned");
+ let directory = cached.upgrade().unwrap_or_else(|| {
+ let temp = tempfile::tempdir().expect("Failed to create golden
warehouse directory");
+ let decoder =
flate2::read::GzDecoder::new(include_bytes!("warehouse.tar.gz").as_slice());
+ tar::Archive::new(decoder)
+ .unpack(temp.path())
+ .expect("Failed to extract golden warehouse");
+ let directory = Arc::new(temp);
+ *cached = Arc::downgrade(&directory);
+ directory
+ });
+ TestWarehouse::Golden(directory)
+}
+
+#[cfg(test)]
+mod tests {
+ use super::{test_warehouse, unpack_warehouse};
+ use std::sync::{Mutex, Weak};
+
+ #[test]
+ fn warehouse_directory_is_removed_on_exit() {
+ const PROBE: &str = "PAIMON_GOLDEN_WAREHOUSE_CLEANUP_PROBE";
+ if let Some(output) = std::env::var_os(PROBE) {
+ let warehouse = test_warehouse();
+ std::fs::write(output, warehouse.path()).unwrap();
+ return;
+ }
+
+ let probe = tempfile::tempdir().unwrap();
+ let output = probe.path().join("warehouse-path");
+ let status =
std::process::Command::new(std::env::current_exe().unwrap())
+ .args([
+ "--exact",
+
"golden_warehouse::tests::warehouse_directory_is_removed_on_exit",
+ ])
+ .env(PROBE, &output)
+ .env_remove("PAIMON_TEST_WAREHOUSE")
+ .status()
+ .unwrap();
+ assert!(status.success());
+ let warehouse = std::fs::read_to_string(output).unwrap();
+ let leaked = std::path::Path::new(&warehouse).exists();
+ if leaked {
+ std::fs::remove_dir_all(&warehouse).unwrap();
+ }
+ assert!(
+ !leaked,
+ "Golden warehouse leaked after process exit: {warehouse}"
+ );
+ }
+ #[test]
+ fn warehouse_lives_until_the_last_reader_finishes() {
+ let cache = Mutex::new(Weak::new());
+ let first = unpack_warehouse(&cache);
+ let second = unpack_warehouse(&cache);
+ assert_eq!(first.path(), second.path());
+ let path = std::path::PathBuf::from(first.path());
+ let schema = path.join("default.db/simple_log_table/schema/schema-0");
+ assert!(schema.is_file());
+ drop(first);
+ assert!(schema.is_file(), "Another reader still owns the warehouse");
+ drop(second);
+ assert!(!path.exists(), "Last reader should clean up the warehouse");
+
+ let next = unpack_warehouse(&cache);
+ assert!(std::path::Path::new(next.path())
+ .join("default.db/simple_log_table/schema/schema-0")
+ .is_file());
+ }
+}
diff --git a/crates/integrations/datafusion/testdata/warehouse/warehouse.tar.gz
b/crates/integrations/datafusion/testdata/warehouse/warehouse.tar.gz
new file mode 100644
index 00000000..42fa10c7
Binary files /dev/null and
b/crates/integrations/datafusion/testdata/warehouse/warehouse.tar.gz differ
diff --git a/crates/integrations/datafusion/tests/dynamic_bucket_tables.rs
b/crates/integrations/datafusion/tests/dynamic_bucket_tables.rs
index 8db284c6..a83db584 100644
--- a/crates/integrations/datafusion/tests/dynamic_bucket_tables.rs
+++ b/crates/integrations/datafusion/tests/dynamic_bucket_tables.rs
@@ -19,6 +19,9 @@
mod common;
+#[path = "../testdata/warehouse/mod.rs"]
+mod golden_warehouse;
+
use common::{
collect_id_name, collect_id_value, create_sql_context, create_test_env,
setup_sql_context,
string_value,
@@ -1047,35 +1050,28 @@ async fn
test_pk_dynamic_bucket_partitioned_insert_overwrite() {
);
}
-/// Read the Spark-provisioned dynamic_bucket_pk_table, write the same data
+/// Read the Java-written golden dynamic_bucket_pk_table, write the same data
/// into a new dynamic bucket table, and verify the HASH index values are
identical.
///
/// Spark may distribute rows across multiple buckets due to parallelism, so we
/// compare the aggregate set of hash values rather than per-bucket entries.
-///
-/// Requires: `make docker-up` + colima copy (see dev/spark/README.md).
#[tokio::test]
-#[ignore]
-async fn test_read_spark_dynamic_bucket_and_compare_index() {
- // --- Read from Spark-provisioned table ---
- let warehouse =
- std::env::var("PAIMON_TEST_WAREHOUSE").unwrap_or_else(|_|
"/tmp/paimon-warehouse".into());
- let mut opts = Options::new();
- opts.set(CatalogOptions::WAREHOUSE, &warehouse);
- let spark_catalog = FileSystemCatalog::new(opts).unwrap();
- let spark_table = spark_catalog
+async fn test_read_golden_dynamic_bucket_and_compare_index() {
+ // Read the Java-written table from the golden warehouse.
+ let (_warehouse, golden_catalog) = golden_warehouse::test_catalog();
+ let golden_table = golden_catalog
.get_table(&Identifier::new("default", "dynamic_bucket_pk_table"))
.await
.unwrap();
- // Read all rows from the Spark table
- let plan = spark_table
+ // Read all rows from the golden table
+ let plan = golden_table
.new_read_builder()
.new_scan()
.plan()
.await
.unwrap();
- let reader = spark_table.new_read_builder().new_read().unwrap();
+ let reader = golden_table.new_read_builder().new_read().unwrap();
let stream = reader.to_arrow(plan.splits()).unwrap();
use arrow_array::RecordBatch;
@@ -1097,7 +1093,7 @@ async fn
test_read_spark_dynamic_bucket_and_compare_index() {
}
all_rows.sort_by_key(|(id, _)| *id);
- // Verify Spark data: two commits with overlapping keys
+ // Verify golden data: two commits with overlapping keys
assert_eq!(
all_rows,
vec![
@@ -1121,7 +1117,7 @@ async fn
test_read_spark_dynamic_bucket_and_compare_index() {
.await
.unwrap();
- // Replicate the same two commits as provision.py
+ // Replicate the same two commits as the golden table
sql_context
.sql("INSERT INTO paimon.test_db.t_dyn_cmp VALUES (1, 'alice'), (2,
'bob'), (3, 'carol')")
.await
@@ -1163,11 +1159,11 @@ async fn
test_read_spark_dynamic_bucket_and_compare_index() {
.await
.unwrap();
- let spark_hashes = collect_all_hashes(&spark_table).await;
+ let golden_hashes = collect_all_hashes(&golden_table).await;
let rust_hashes = collect_all_hashes(&rust_table).await;
assert_eq!(
- spark_hashes, rust_hashes,
- "HASH index values mismatch between Spark and Rust tables"
+ golden_hashes, rust_hashes,
+ "HASH index values mismatch between the golden and Rust tables"
);
}
diff --git a/crates/integrations/datafusion/tests/read_tables.rs
b/crates/integrations/datafusion/tests/read_tables.rs
index ef42a2d4..dbd626cf 100644
--- a/crates/integrations/datafusion/tests/read_tables.rs
+++ b/crates/integrations/datafusion/tests/read_tables.rs
@@ -20,6 +20,10 @@ use std::sync::Arc;
mod common;
+#[path = "../testdata/warehouse/mod.rs"]
+mod golden_warehouse;
+use golden_warehouse::{test_catalog as create_catalog, TestWarehouse};
+
use common::string_value;
use datafusion::arrow::array::{Array, Int32Array};
use datafusion::arrow::record_batch::RecordBatch;
@@ -30,46 +34,28 @@ use datafusion::logical_expr::{col, lit,
TableProviderFilterPushDown};
use datafusion::physical_plan::{displayable, ExecutionPlan};
use datafusion::prelude::SessionConfig;
use paimon::catalog::Identifier;
-use paimon::{Catalog, CatalogOptions, FileSystemCatalog, Options};
+use paimon::Catalog;
use paimon_datafusion::{PaimonCatalogProvider, PaimonTableProvider,
SQLContext};
-fn get_test_warehouse() -> String {
- std::env::var("PAIMON_TEST_WAREHOUSE").unwrap_or_else(|_|
"/tmp/paimon-warehouse".to_string())
-}
-
-fn create_catalog() -> FileSystemCatalog {
- let warehouse = get_test_warehouse();
- let mut options = Options::new();
- options.set(CatalogOptions::WAREHOUSE, warehouse);
- FileSystemCatalog::new(options).expect("Failed to create catalog")
-}
-
-async fn create_context() -> SQLContext {
- let catalog = create_catalog();
+async fn create_context() -> (TestWarehouse, SQLContext) {
+ let (warehouse, catalog) = create_catalog();
let catalog: Arc<dyn Catalog> = Arc::new(catalog);
let mut ctx = SQLContext::new();
ctx.register_catalog("paimon", catalog)
.await
.expect("Failed to register catalog");
- ctx
+ (warehouse, ctx)
}
-async fn create_provider(table_name: &str) -> PaimonTableProvider {
- let catalog = create_catalog();
- let identifier = Identifier::new("default", table_name);
- let table = catalog
- .get_table(&identifier)
- .await
- .expect("Failed to get table");
-
- PaimonTableProvider::try_new(table).expect("Failed to create table
provider")
+async fn create_provider(table_name: &str) -> (TestWarehouse,
PaimonTableProvider) {
+ create_provider_with_options(table_name, HashMap::new()).await
}
async fn create_provider_with_options(
table_name: &str,
extra_options: HashMap<String, String>,
-) -> PaimonTableProvider {
- let catalog = create_catalog();
+) -> (TestWarehouse, PaimonTableProvider) {
+ let (warehouse, catalog) = create_catalog();
let identifier = Identifier::new("default", table_name);
let table = catalog
.get_table(&identifier)
@@ -77,7 +63,8 @@ async fn create_provider_with_options(
.expect("Failed to get table")
.copy_with_options(extra_options);
- PaimonTableProvider::try_new(table).expect("Failed to create table
provider")
+ let provider = PaimonTableProvider::try_new(table).expect("Failed to
create table provider");
+ (warehouse, provider)
}
async fn read_rows(table_name: &str) -> Vec<(i32, String)> {
@@ -97,7 +84,7 @@ async fn read_rows(table_name: &str) -> Vec<(i32, String)> {
}
async fn collect_query(sql: &str) ->
datafusion::error::Result<Vec<RecordBatch>> {
- let ctx = create_context().await;
+ let (_warehouse, ctx) = create_context().await;
ctx.sql(sql).await?.collect().await
}
@@ -149,9 +136,12 @@ async fn assert_time_travel_schema_evolution_rows(
assert_sql_rows(&sql, expected).await;
}
-async fn create_physical_plan(sql: &str) -> datafusion::error::Result<Arc<dyn
ExecutionPlan>> {
- let ctx = create_context().await;
- ctx.sql(sql).await?.create_physical_plan().await
+async fn create_physical_plan(
+ sql: &str,
+) -> datafusion::error::Result<(TestWarehouse, Arc<dyn ExecutionPlan>)> {
+ let (warehouse, ctx) = create_context().await;
+ let plan = ctx.sql(sql).await?.create_physical_plan().await?;
+ Ok((warehouse, plan))
}
fn extract_id_name_rows(
@@ -261,7 +251,7 @@ async fn test_projection_via_datafusion() {
#[tokio::test]
async fn test_supports_partition_filters_pushdown() {
- let provider = create_provider("multi_partitioned_log_table").await;
+ let (_warehouse, provider) =
create_provider("multi_partitioned_log_table").await;
let partition_filter = col("dt").eq(lit("2024-01-01"));
let mixed_and_filter =
col("dt").eq(lit("2024-01-01")).and(col("id").gt(lit(1)));
let data_filter = col("id").gt(lit(1));
@@ -285,7 +275,7 @@ async fn test_supports_partition_filters_pushdown() {
/// partition count is still capped by `target_partitions`.
#[tokio::test]
async fn test_scan_partition_count_respects_session_config() {
- let provider = create_provider("partitioned_log_table").await;
+ let (_warehouse, provider) =
create_provider("partitioned_log_table").await;
// With generous target_partitions, the plan should expose more than one
partition.
let config = SessionConfig::new().with_target_partitions(8);
@@ -369,7 +359,7 @@ async fn
test_mixed_and_filter_keeps_residual_datafusion_filter() {
#[tokio::test]
async fn
test_partially_translated_filter_keeps_partition_pruning_and_correctness() {
let sql = "SELECT id, name FROM paimon.default.multi_partitioned_log_table
WHERE dt = '2024-01-01' AND hr + 1 > 20 LIMIT 1";
- let plan = create_physical_plan(sql)
+ let (_warehouse, plan) = create_physical_plan(sql)
.await
.expect("Physical plan creation should succeed");
let plan_text = format_physical_plan(&plan);
@@ -621,7 +611,7 @@ async fn
test_limit_pushdown_on_data_evolution_table_returns_merged_rows() {
#[tokio::test]
async fn
test_limit_pushdown_marks_safe_scan_limit_hint_and_keeps_correctness() {
let sql = "SELECT id, name FROM paimon.default.simple_log_table LIMIT 2";
- let plan = create_physical_plan(sql)
+ let (_warehouse, plan) = create_physical_plan(sql)
.await
.expect("Physical plan creation should succeed");
let plan_text = format_physical_plan(&plan);
@@ -644,7 +634,7 @@ async fn
test_limit_pushdown_marks_safe_scan_limit_hint_and_keeps_correctness()
#[tokio::test]
async fn test_offset_limit_pushdown_keeps_correctness_without_fetch_contract()
{
let sql = "SELECT id, name FROM paimon.default.partitioned_log_table
OFFSET 1 LIMIT 1";
- let plan = create_physical_plan(sql)
+ let (_warehouse, plan) = create_physical_plan(sql)
.await
.expect("Physical plan creation should succeed");
let plan_text = format_physical_plan(&plan);
@@ -673,7 +663,7 @@ async fn
test_offset_limit_pushdown_keeps_correctness_without_fetch_contract() {
#[tokio::test]
async fn test_inexact_filter_limit_keeps_correctness_without_fetch_contract() {
let sql = "SELECT id, name FROM paimon.default.partitioned_log_table WHERE
id > 1 LIMIT 1";
- let plan = create_physical_plan(sql)
+ let (_warehouse, plan) = create_physical_plan(sql)
.await
.expect("Physical plan creation should succeed");
let plan_text = format_physical_plan(&plan);
@@ -701,7 +691,7 @@ async fn
test_inexact_filter_limit_keeps_correctness_without_fetch_contract() {
#[tokio::test]
async fn test_residual_filter_limit_keeps_connector_limit_and_correctness() {
let sql = "SELECT id, name FROM paimon.default.simple_log_table WHERE id +
1 > 3 LIMIT 1";
- let plan = create_physical_plan(sql)
+ let (_warehouse, plan) = create_physical_plan(sql)
.await
.expect("Physical plan creation should succeed");
let plan_text = format_physical_plan(&plan);
@@ -737,7 +727,7 @@ async fn
test_residual_filter_limit_keeps_connector_limit_and_correctness() {
// ======================= Catalog Provider Tests =======================
#[tokio::test]
async fn test_query_via_catalog_provider() {
- let catalog = create_catalog();
+ let (_warehouse, catalog) = create_catalog();
let catalog: Arc<dyn Catalog> = Arc::new(catalog);
let mut ctx = SQLContext::new();
ctx.register_catalog("paimon", catalog)
@@ -756,7 +746,7 @@ async fn test_query_via_catalog_provider() {
#[tokio::test]
async fn test_missing_database_returns_no_schema() {
- let catalog = create_catalog();
+ let (_warehouse, catalog) = create_catalog();
let provider = PaimonCatalogProvider::new(
None,
Arc::new(catalog),
@@ -773,20 +763,9 @@ async fn test_missing_database_returns_no_schema() {
// ======================= Time Travel Tests =======================
-/// Helper: create a SQLContext with catalog + relation planner for time
travel.
-async fn create_time_travel_context() -> SQLContext {
- let catalog = create_catalog();
- let catalog: Arc<dyn Catalog> = Arc::new(catalog);
- let mut ctx = SQLContext::new();
- ctx.register_catalog("paimon", catalog)
- .await
- .expect("Failed to register catalog");
- ctx
-}
-
#[tokio::test]
async fn test_time_travel_by_snapshot_id() {
- let ctx = create_time_travel_context().await;
+ let (_warehouse, ctx) = create_context().await;
// Snapshot 1: should contain only the first insert (alice, bob)
let batches = ctx
@@ -830,7 +809,7 @@ async fn test_time_travel_by_snapshot_id() {
#[tokio::test]
async fn test_time_travel_by_tag_name() {
- let ctx = create_time_travel_context().await;
+ let (_warehouse, ctx) = create_context().await;
// Tag 'snapshot1' points to snapshot 1: should contain only (alice, bob)
let batches = ctx
@@ -1014,7 +993,7 @@ async fn test_time_travel_conflicting_selectors_fail() {
("1", "scan.snapshot-id"),
("watermark-1", "scan.watermark"),
] {
- let provider = create_provider_with_options(
+ let (_provider_warehouse, provider) = create_provider_with_options(
"time_travel_table",
HashMap::from([
("scan.version".to_string(), version.to_string()),
@@ -1023,7 +1002,7 @@ async fn test_time_travel_conflicting_selectors_fail() {
)
.await;
- let ctx = create_context().await;
+ let (_warehouse, ctx) = create_context().await;
ctx.register_temp_table("paimon.default.time_travel_table",
Arc::new(provider))
.expect("Failed to register temp table");
@@ -1051,13 +1030,13 @@ async fn test_time_travel_conflicting_selectors_fail() {
#[tokio::test]
async fn test_time_travel_invalid_version_fails() {
- let provider = create_provider_with_options(
+ let (_provider_warehouse, provider) = create_provider_with_options(
"time_travel_table",
HashMap::from([("scan.version".to_string(),
"nonexistent-tag".to_string())]),
)
.await;
- let ctx = create_context().await;
+ let (_warehouse, ctx) = create_context().await;
ctx.register_temp_table("paimon.default.time_travel_table",
Arc::new(provider))
.expect("Failed to register temp table");
@@ -1349,7 +1328,7 @@ async fn test_read_complex_type_table_via_datafusion() {
async fn test_select_row_id_from_data_evolution_table() {
use datafusion::arrow::array::Int64Array;
- let ctx = create_context().await;
+ let (_warehouse, ctx) = create_context().await;
let batches = ctx
.sql(r#"SELECT "_ROW_ID", id, name FROM
paimon.default.data_evolution_table"#)
@@ -1385,7 +1364,7 @@ async fn test_select_row_id_from_data_evolution_table() {
async fn test_filter_row_id_from_data_evolution_table() {
use datafusion::arrow::array::Int64Array;
- let ctx = create_context().await;
+ let (_warehouse, ctx) = create_context().await;
let all_batches = ctx
.sql(r#"SELECT "_ROW_ID" FROM paimon.default.data_evolution_table"#)
diff --git a/dev/docker-compose.yaml b/dev/docker-compose.yaml
deleted file mode 100644
index c5431c09..00000000
--- a/dev/docker-compose.yaml
+++ /dev/null
@@ -1,41 +0,0 @@
-# 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.
-#
-# Spark + Paimon for integration tests: writes tables to a local volume
-# so paimon-rust can read them (no MinIO/S3).
-#
-# Usage:
-# 1. Generate test data (run once):
-# docker compose -f dev/docker-compose.yaml run --rm spark-paimon
-# 2. Run read integration tests on the host:
-# PAIMON_TEST_WAREHOUSE=/tmp/paimon-warehouse cargo test -p
paimon-integration-tests read_from_spark
-#
-# To re-provision: remove /tmp/paimon-warehouse and run step 1 again.
-
-services:
- spark-paimon:
- build:
- context: ./spark
- dockerfile: Dockerfile
- volumes:
- # Bind warehouse to host so paimon-rust can read it
- - /tmp/paimon-warehouse:/tmp/paimon-warehouse
- # Run as root to avoid permission issues with mounted volumes
- user: root
- # Run provision script and exit (data remains in /tmp/paimon-warehouse).
- # Use full path: entrypoint exec does not have SPARK_HOME/bin in PATH.
- command: ["/opt/spark/bin/spark-submit", "--master", "local[*]",
"/opt/provision.py"]
diff --git a/dev/spark/Dockerfile b/dev/spark/Dockerfile
deleted file mode 100644
index baf0d2bc..00000000
--- a/dev/spark/Dockerfile
+++ /dev/null
@@ -1,53 +0,0 @@
-# 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.
-
-# Spark with Paimon connector for integration tests.
-# Writes Paimon tables to /tmp/paimon-warehouse (mount this as a volume
-# so paimon-rust can read the data).
-#
-# Build args can be overridden via Makefile or docker build:
-# --build-arg PAIMON_VERSION=1.3.1
-FROM apache/spark:3.5.3
-
-# Re-declare ARGs so they are available in this stage (ARGs before FROM are
not).
-ARG PAIMON_VERSION
-
-USER root
-WORKDIR ${SPARK_HOME}
-
-RUN apt-get install -y --no-install-recommends curl && \
- rm -rf /var/lib/apt/lists/*
-
-# Paimon Spark runtime JAR (Spark 3.5)
-#
https://repo.maven.apache.org/maven2/org/apache/paimon/paimon-spark-3.5/1.3.1/paimon-spark-3.5-1.3.1.jar
-ENV PAIMON_JAR="paimon-spark-3.5-${PAIMON_VERSION}.jar"
-ENV
PAIMON_JAR_URL="https://repo.maven.apache.org/maven2/org/apache/paimon/paimon-spark-3.5/${PAIMON_VERSION}/${PAIMON_JAR}";
-
-RUN curl -fsSL --retry 3 -o "${SPARK_HOME}/jars/${PAIMON_JAR}" \
- "${PAIMON_JAR_URL}" && \
- chown -R spark:spark "${SPARK_HOME}/jars"
-
-COPY spark-defaults.conf ${SPARK_HOME}/conf/
-COPY provision.py /opt/provision.py
-RUN chown spark:spark ${SPARK_HOME}/conf/spark-defaults.conf /opt/provision.py
-
-USER spark
-
-# Default: run provision script to create tables and write data to
/tmp/paimon-warehouse.
-# Override in docker-compose or when running manually.
-# Use full path: entrypoint exec does not have SPARK_HOME/bin in PATH.
-CMD ["/opt/spark/bin/spark-submit", "--master", "local[*]",
"/opt/provision.py"]
diff --git a/dev/spark/README.md b/dev/spark/README.md
deleted file mode 100644
index 519a526f..00000000
--- a/dev/spark/README.md
+++ /dev/null
@@ -1,60 +0,0 @@
-<!--
-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.
--->
-
-# Spark Provisioning for Integration Tests
-
-This directory contains the Spark + Paimon setup that provisions test tables
into `/tmp/paimon-warehouse`.
-
-## Prerequisites
-
-- Docker (via colima or Docker Desktop)
-- `docker compose`
-
-## Provision Test Data
-
-```bash
-# Build and run (from repo root):
-make docker-up
-
-# Or manually:
-docker compose -f dev/docker-compose.yaml build --build-arg
PAIMON_VERSION=1.3.1
-docker compose -f dev/docker-compose.yaml run --rm spark-paimon
-```
-
-`provision.py` automatically clears the warehouse directory before creating
tables, so re-running is always safe.
-
-### colima Users
-
-With colima, Docker volumes mount inside the colima VM, **not** on the macOS
host filesystem. After provisioning, copy the data to the host:
-
-```bash
-colima ssh -- sudo tar cf - -C /tmp paimon-warehouse | tar xf - -C /tmp
-```
-
-## Run Integration Tests
-
-```bash
-cargo test -p paimon-integration-tests
-```
-
-## Files
-
-- `Dockerfile` — Spark 3.5 + Paimon connector image
-- `spark-defaults.conf` — Spark config with Paimon catalog
-- `provision.py` — Creates all test tables and inserts data
diff --git a/dev/spark/provision.py b/dev/spark/provision.py
deleted file mode 100644
index 0b209d0e..00000000
--- a/dev/spark/provision.py
+++ /dev/null
@@ -1,1363 +0,0 @@
-#!/usr/bin/env python3
-#
-# 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.
-#
-# Provisions Paimon tables into the warehouse (file:/tmp/paimon-warehouse)
-# for paimon-rust integration tests to read.
-
-import shutil
-import time
-from pathlib import Path
-from urllib.parse import unquote, urlparse
-
-from pyspark.sql import SparkSession
-
-
-def _warehouse_path_from_spark_conf(spark: SparkSession) -> Path:
- warehouse_uri = spark.conf.get("spark.sql.catalog.paimon.warehouse")
- parsed = urlparse(warehouse_uri)
-
- if parsed.scheme not in ("", "file"):
- raise ValueError(
- f"Unsupported Paimon warehouse URI scheme {parsed.scheme!r}:
{warehouse_uri}"
- )
-
- if parsed.netloc not in ("", "localhost"):
- raise ValueError(
- f"Unsupported remote Paimon warehouse location {parsed.netloc!r}:
{warehouse_uri}"
- )
-
- warehouse_path = Path(unquote(parsed.path if parsed.scheme else
warehouse_uri))
- if not warehouse_path.is_absolute() or str(warehouse_path) == "/":
- raise ValueError(f"Refusing to clear unsafe warehouse path:
{warehouse_path}")
-
- return warehouse_path
-
-
-def _reset_warehouse_dir(warehouse_path: Path) -> None:
- warehouse_path.mkdir(parents=True, exist_ok=True)
-
- for child in warehouse_path.iterdir():
- for attempt in range(3):
- try:
- if child.is_symlink() or child.is_file():
- child.unlink()
- else:
- shutil.rmtree(child)
- break
- except OSError:
- if attempt == 2:
- raise
- time.sleep(0.1)
-
-
-def main():
- spark = SparkSession.builder.getOrCreate()
-
- warehouse_path = _warehouse_path_from_spark_conf(spark)
- _reset_warehouse_dir(warehouse_path)
-
- # Use Paimon catalog (configured in spark-defaults.conf with warehouse
file:/tmp/paimon-warehouse)
- spark.sql("USE paimon.default")
-
- # Table: simple log table for read tests
- spark.sql(
- """
- CREATE TABLE IF NOT EXISTS simple_log_table (
- id INT,
- name STRING
- ) USING paimon
- """
- )
- spark.sql("INSERT INTO simple_log_table VALUES (1, 'alice'), (2, 'bob'),
(3, 'carol')")
-
- # Spark SQL here does not accept table constraints like
- # PRIMARY KEY (id) NOT ENFORCED inside the column list, so use
- # Paimon table properties instead.
- spark.sql(
- """
- CREATE TABLE IF NOT EXISTS simple_pk_table (
- id INT,
- name STRING
- ) USING paimon
- TBLPROPERTIES (
- 'primary-key' = 'id',
- 'bucket' = '1'
- )
- """
- )
- spark.sql(
- """
- INSERT INTO simple_pk_table VALUES
- (1, 'alice'),
- (2, 'bob'),
- (3, 'carol')
- """
- )
-
- # Table: primary key table with deletion vectors enabled.
- # Re-inserting the same keys with newer values creates deleted historical
- # rows that readers must filter via deletion vectors.
- spark.sql(
- """
- CREATE TABLE IF NOT EXISTS simple_dv_pk_table (
- id INT,
- name STRING
- ) USING paimon
- TBLPROPERTIES (
- 'primary-key' = 'id',
- 'bucket' = '2',
- 'deletion-vectors.enabled' = 'true'
- )
- """
- )
-
- spark.sql(
- """
- INSERT INTO simple_dv_pk_table VALUES
- (1, 'alice-v1'),
- (2, 'bob-v1'),
- (3, 'carol-v1'),
- (5, 'eve-v1')
- """
- )
-
- spark.sql(
- """
- INSERT INTO simple_dv_pk_table VALUES
- (2, 'bob-v2'),
- (3, 'carol-v2'),
- (4, 'dave-v1'),
- (6, 'frank-v1')
- """
- )
-
- spark.sql(
- """
- INSERT INTO simple_dv_pk_table VALUES
- (1, 'alice-v2'),
- (4, 'dave-v2'),
- (5, 'eve-v2')
- """
- )
-
- # ===== Partitioned table: single partition key (dt) =====
- spark.sql(
- """
- CREATE TABLE IF NOT EXISTS partitioned_log_table (
- id INT,
- name STRING,
- dt STRING
- ) USING paimon
- PARTITIONED BY (dt)
- """
- )
- spark.sql(
- """
- INSERT INTO partitioned_log_table VALUES
- (1, 'alice', '2024-01-01'),
- (2, 'bob', '2024-01-01'),
- (3, 'carol', '2024-01-02')
- """
- )
-
- # ===== Partitioned table: multiple partition keys (dt, hr) =====
- spark.sql(
- """
- CREATE TABLE IF NOT EXISTS multi_partitioned_log_table (
- id INT,
- name STRING,
- dt STRING,
- hr INT
- ) USING paimon
- PARTITIONED BY (dt, hr)
- """
- )
- spark.sql(
- """
- INSERT INTO multi_partitioned_log_table VALUES
- (1, 'alice', '2024-01-01', 10),
- (2, 'bob', '2024-01-01', 10),
- (3, 'carol', '2024-01-01', 20),
- (4, 'dave', '2024-01-02', 10)
- """
- )
-
- # ===== Partitioned table: PK + DV enabled =====
- spark.sql(
- """
- CREATE TABLE IF NOT EXISTS partitioned_dv_pk_table (
- id INT,
- name STRING,
- dt STRING
- ) USING paimon
- PARTITIONED BY (dt)
- TBLPROPERTIES (
- 'primary-key' = 'id,dt',
- 'bucket' = '1',
- 'deletion-vectors.enabled' = 'true'
- )
- """
- )
-
- spark.sql(
- """
- INSERT INTO partitioned_dv_pk_table VALUES
- (1, 'alice-v1', '2024-01-01'),
- (2, 'bob-v1', '2024-01-01'),
- (1, 'alice-v1', '2024-01-02'),
- (3, 'carol-v1', '2024-01-02')
- """
- )
-
- spark.sql(
- """
- INSERT INTO partitioned_dv_pk_table VALUES
- (1, 'alice-v2', '2024-01-01'),
- (3, 'carol-v2', '2024-01-02'),
- (4, 'dave-v1', '2024-01-02')
- """
- )
-
- spark.sql(
- """
- INSERT INTO partitioned_dv_pk_table VALUES
- (2, 'bob-v2', '2024-01-01'),
- (4, 'dave-v2', '2024-01-02')
- """
- )
-
- # ===== Data Evolution table: append-only with row tracking =====
- # data-evolution.enabled + row-tracking.enabled allows partial column
updates
- # via MERGE INTO. This produces files with different write_cols covering
the
- # same row ID ranges, exercising the column-wise merge read path.
- spark.sql(
- """
- CREATE TABLE IF NOT EXISTS data_evolution_table (
- id INT,
- name STRING,
- value INT
- ) USING paimon
- TBLPROPERTIES (
- 'row-tracking.enabled' = 'true',
- 'data-evolution.enabled' = 'true'
- )
- """
- )
-
- # First batch: rows with row_id 0, 1, 2 — all columns written
- spark.sql(
- """
- INSERT INTO data_evolution_table VALUES
- (1, 'alice', 100),
- (2, 'bob', 200),
- (3, 'carol', 300)
- """
- )
-
- # Second batch: rows with row_id 3, 4
- spark.sql(
- """
- INSERT INTO data_evolution_table VALUES
- (4, 'dave', 400),
- (5, 'eve', 500)
- """
- )
-
- # MERGE INTO: partial column update on existing rows.
- # This writes new files containing only the updated column (name) with the
- # same first_row_id, so the reader must merge columns from multiple files.
- # Paimon 1.3.1 requires the source table to be a Paimon table.
- spark.sql(
- """
- CREATE TABLE IF NOT EXISTS data_evolution_updates (
- id INT,
- name STRING
- ) USING paimon
- """
- )
- spark.sql(
- """
- INSERT INTO data_evolution_updates VALUES (1, 'alice-v2'), (3,
'carol-v2')
- """
- )
- spark.sql(
- """
- MERGE INTO data_evolution_table t
- USING data_evolution_updates s
- ON t.id = s.id
- WHEN MATCHED THEN UPDATE SET t.name = s.name
- """
- )
- spark.sql("DROP TABLE data_evolution_updates")
-
- # ===== Time travel table: multiple snapshots for time travel tests =====
- # Snapshot 1: rows (1, 'alice'), (2, 'bob')
- # Snapshot 2: rows (1, 'alice'), (2, 'bob'), (3, 'carol'), (4, 'dave')
- spark.sql(
- """
- CREATE TABLE IF NOT EXISTS time_travel_table (
- id INT,
- name STRING
- ) USING paimon
- """
- )
- spark.sql(
- """
- INSERT INTO time_travel_table VALUES
- (1, 'alice'),
- (2, 'bob')
- """
- )
- spark.sql(
- """
- INSERT INTO time_travel_table VALUES
- (3, 'carol'),
- (4, 'dave')
- """
- )
-
- # Create tags for tag-based time travel tests
- # Tag 'snapshot1' points to snapshot 1 (alice, bob)
- # Tag 'snapshot2' points to snapshot 2 (alice, bob, carol, dave)
- spark.sql("CALL sys.create_tag('default.time_travel_table', 'snapshot1',
1)")
- spark.sql("CALL sys.create_tag('default.time_travel_table', 'snapshot2',
2)")
-
- # ===== Time travel + schema evolution table =====
- # Each tag points at a stable schema boundary. Tests read by tag name
rather
- # than depending on global warehouse snapshot numbering.
- spark.sql(
- """
- CREATE TABLE IF NOT EXISTS time_travel_schema_evolution (
- id INT,
- name STRING
- ) USING paimon
- """
- )
- spark.sql(
- """
- INSERT INTO time_travel_schema_evolution VALUES
- (1, 'alice'),
- (2, 'bob')
- """
- )
- spark.sql("CALL sys.create_tag('default.time_travel_schema_evolution',
'before_add_column', 1)")
-
- spark.sql("ALTER TABLE time_travel_schema_evolution ADD COLUMNS (age INT)")
- spark.sql(
- """
- INSERT INTO time_travel_schema_evolution VALUES
- (3, 'carol', 30),
- (4, 'dave', 40)
- """
- )
- spark.sql("CALL sys.create_tag('default.time_travel_schema_evolution',
'after_add_column', 2)")
- spark.sql("CALL sys.create_tag('default.time_travel_schema_evolution',
'before_rename', 2)")
-
- spark.sql("ALTER TABLE time_travel_schema_evolution RENAME COLUMN name TO
full_name")
- spark.sql(
- """
- INSERT INTO time_travel_schema_evolution VALUES
- (5, 'erin', 50),
- (6, 'frank', 60)
- """
- )
- spark.sql("CALL sys.create_tag('default.time_travel_schema_evolution',
'after_rename', 3)")
- spark.sql("CALL sys.create_tag('default.time_travel_schema_evolution',
'before_drop', 3)")
-
- spark.sql("ALTER TABLE time_travel_schema_evolution DROP COLUMN age")
- spark.sql(
- """
- INSERT INTO time_travel_schema_evolution VALUES
- (7, 'grace'),
- (8, 'hank')
- """
- )
- spark.sql("CALL sys.create_tag('default.time_travel_schema_evolution',
'after_drop', 4)")
- spark.sql("CALL sys.create_tag('default.time_travel_schema_evolution',
'before_reorder', 4)")
-
- spark.sql("ALTER TABLE time_travel_schema_evolution ALTER COLUMN full_name
FIRST")
- spark.sql(
- """
- INSERT INTO time_travel_schema_evolution VALUES
- ('ivy', 9),
- ('jane', 10)
- """
- )
- spark.sql("CALL sys.create_tag('default.time_travel_schema_evolution',
'after_reorder', 5)")
-
- # ===== Schema Evolution: Add Column =====
- # Old files have (id, name); after ALTER TABLE ADD COLUMNS, new files have
(id, name, age).
- # Reader must fill nulls for 'age' when reading old files.
- spark.sql(
- """
- CREATE TABLE IF NOT EXISTS schema_evolution_add_column (
- id INT,
- name STRING
- ) USING paimon
- """
- )
- spark.sql(
- "INSERT INTO schema_evolution_add_column VALUES (1, 'alice'), (2,
'bob')"
- )
- spark.sql("ALTER TABLE schema_evolution_add_column ADD COLUMNS (age INT)")
- spark.sql(
- "INSERT INTO schema_evolution_add_column VALUES (3, 'carol', 30), (4,
'dave', 40)"
- )
-
- # ===== Schema Evolution: Type Promotion (INT -> BIGINT) =====
- # Old files have value as INT; after ALTER TABLE, new files have value as
BIGINT.
- # Reader must cast INT to BIGINT when reading old files.
- spark.sql(
- """
- CREATE TABLE IF NOT EXISTS schema_evolution_type_promotion (
- id INT,
- value INT
- ) USING paimon
- """
- )
- spark.sql(
- "INSERT INTO schema_evolution_type_promotion VALUES (1, 100), (2, 200)"
- )
- spark.sql(
- "ALTER TABLE schema_evolution_type_promotion ALTER COLUMN value TYPE
BIGINT"
- )
- spark.sql(
- "INSERT INTO schema_evolution_type_promotion VALUES (3, 3000000000)"
- )
-
- # ===== Mixed-format Schema Evolution: Add Column =====
- # Old Parquet files lack age; new ORC/Avro files contain age.
- spark.sql(
- """
- CREATE TABLE IF NOT EXISTS format_schema_evolution_add_column (
- id INT,
- name STRING
- ) USING paimon
- TBLPROPERTIES (
- 'file.format' = 'parquet'
- )
- """
- )
- spark.sql(
- "INSERT INTO format_schema_evolution_add_column VALUES (1, 'alice'),
(2, 'bob')"
- )
- spark.sql("ALTER TABLE format_schema_evolution_add_column ADD COLUMNS (age
INT)")
- spark.sql("ALTER TABLE format_schema_evolution_add_column SET
TBLPROPERTIES ('file.format' = 'orc')")
- spark.sql(
- "INSERT INTO format_schema_evolution_add_column VALUES (3, 'carol',
30), (4, 'dave', 40)"
- )
- spark.sql("ALTER TABLE format_schema_evolution_add_column SET
TBLPROPERTIES ('file.format' = 'avro')")
- spark.sql(
- "INSERT INTO format_schema_evolution_add_column VALUES (5, 'eve', 50),
(6, 'frank', 60)"
- )
-
- # ===== Partitioned Mixed-format Schema Evolution: Add Column =====
- # Old Parquet files lack extra; new ORC/Avro files contain extra across dt
partitions.
- spark.sql(
- """
- CREATE TABLE IF NOT EXISTS
partitioned_format_schema_evolution_add_column (
- id INT,
- name STRING,
- dt STRING
- ) USING paimon
- PARTITIONED BY (dt)
- TBLPROPERTIES (
- 'file.format' = 'parquet'
- )
- """
- )
- spark.sql(
- """
- INSERT INTO partitioned_format_schema_evolution_add_column VALUES
- (1, 'alice', '2024-01-01'),
- (2, 'bob', '2024-01-02')
- """
- )
- spark.sql(
- "ALTER TABLE partitioned_format_schema_evolution_add_column ADD
COLUMNS (extra STRING)"
- )
- spark.sql(
- "ALTER TABLE partitioned_format_schema_evolution_add_column SET
TBLPROPERTIES ('file.format' = 'orc')"
- )
- spark.sql(
- """
- INSERT INTO partitioned_format_schema_evolution_add_column (id, name,
extra, dt) VALUES
- (3, 'carol', 'orc-extra-1', '2024-01-01'),
- (4, 'dave', 'orc-extra-2', '2024-01-03')
- """
- )
- spark.sql(
- "ALTER TABLE partitioned_format_schema_evolution_add_column SET
TBLPROPERTIES ('file.format' = 'avro')"
- )
- spark.sql(
- """
- INSERT INTO partitioned_format_schema_evolution_add_column (id, name,
extra, dt) VALUES
- (5, 'eve', 'avro-extra-1', '2024-01-02'),
- (6, 'frank', 'avro-extra-2', '2024-01-03')
- """
- )
-
- # ===== Mixed-format Schema Evolution: Type Promotion (INT -> BIGINT) =====
- # Old Parquet files have value as INT; new ORC/Avro files have value as
BIGINT.
- spark.sql(
- """
- CREATE TABLE IF NOT EXISTS format_schema_evolution_type_promotion (
- id INT,
- value INT
- ) USING paimon
- TBLPROPERTIES (
- 'file.format' = 'parquet'
- )
- """
- )
- spark.sql(
- "INSERT INTO format_schema_evolution_type_promotion VALUES (1, 100),
(2, 200)"
- )
- spark.sql(
- "ALTER TABLE format_schema_evolution_type_promotion ALTER COLUMN value
TYPE BIGINT"
- )
- spark.sql("ALTER TABLE format_schema_evolution_type_promotion SET
TBLPROPERTIES ('file.format' = 'orc')")
- spark.sql(
- "INSERT INTO format_schema_evolution_type_promotion VALUES (3,
3000000000), (4, 4000000000)"
- )
- spark.sql("ALTER TABLE format_schema_evolution_type_promotion SET
TBLPROPERTIES ('file.format' = 'avro')")
- spark.sql(
- "INSERT INTO format_schema_evolution_type_promotion VALUES (5,
5000000000), (6, 6000000000)"
- )
-
- # ===== Mixed-format Data Evolution: Add Column =====
- # Combines row-tracking/data-evolution with ADD COLUMN and mixed file
formats.
- # Old Parquet files lack extra; new ORC/Avro files contain extra.
- spark.sql(
- """
- CREATE TABLE IF NOT EXISTS data_evolution_mixed_format_add_column (
- id INT,
- name STRING,
- value INT
- ) USING paimon
- TBLPROPERTIES (
- 'row-tracking.enabled' = 'true',
- 'data-evolution.enabled' = 'true',
- 'file.format' = 'parquet'
- )
- """
- )
- spark.sql(
- """
- INSERT INTO data_evolution_mixed_format_add_column VALUES
- (1, 'alice', 100),
- (2, 'bob', 200)
- """
- )
- spark.sql("ALTER TABLE data_evolution_mixed_format_add_column ADD COLUMNS
(extra STRING)")
- spark.sql("ALTER TABLE data_evolution_mixed_format_add_column SET
TBLPROPERTIES ('file.format' = 'orc')")
- spark.sql(
- "INSERT INTO data_evolution_mixed_format_add_column VALUES (3,
'carol', 300, 'orc-extra')"
- )
- spark.sql("ALTER TABLE data_evolution_mixed_format_add_column SET
TBLPROPERTIES ('file.format' = 'avro')")
- spark.sql(
- "INSERT INTO data_evolution_mixed_format_add_column VALUES (4, 'dave',
400, 'avro-extra')"
- )
-
- # ===== Mixed-format Data Evolution: Type Promotion =====
- # Old Parquet files have INT; new ORC/Avro files have BIGINT.
- spark.sql(
- """
- CREATE TABLE IF NOT EXISTS data_evolution_mixed_format_type_promotion (
- id INT,
- value INT
- ) USING paimon
- TBLPROPERTIES (
- 'row-tracking.enabled' = 'true',
- 'data-evolution.enabled' = 'true',
- 'file.format' = 'parquet'
- )
- """
- )
- spark.sql(
- "INSERT INTO data_evolution_mixed_format_type_promotion VALUES (1,
100), (2, 200)"
- )
- spark.sql(
- "ALTER TABLE data_evolution_mixed_format_type_promotion ALTER COLUMN
value TYPE BIGINT"
- )
- spark.sql("ALTER TABLE data_evolution_mixed_format_type_promotion SET
TBLPROPERTIES ('file.format' = 'orc')")
- spark.sql(
- "INSERT INTO data_evolution_mixed_format_type_promotion VALUES (3,
3000000000)"
- )
- spark.sql("ALTER TABLE data_evolution_mixed_format_type_promotion SET
TBLPROPERTIES ('file.format' = 'avro')")
- spark.sql(
- "INSERT INTO data_evolution_mixed_format_type_promotion VALUES (4,
4000000000)"
- )
-
- # ===== Data Evolution + Schema Evolution: Add Column =====
- # Combines data-evolution (row-tracking + MERGE INTO) with ALTER TABLE ADD
COLUMNS.
- # Old files lack the new column; MERGE INTO produces partial-column files.
- # Reader must fill nulls for missing columns AND merge columns across
files.
- spark.sql(
- """
- CREATE TABLE IF NOT EXISTS data_evolution_add_column (
- id INT,
- name STRING,
- value INT
- ) USING paimon
- TBLPROPERTIES (
- 'row-tracking.enabled' = 'true',
- 'data-evolution.enabled' = 'true'
- )
- """
- )
- spark.sql(
- """
- INSERT INTO data_evolution_add_column VALUES
- (1, 'alice', 100),
- (2, 'bob', 200)
- """
- )
- spark.sql("ALTER TABLE data_evolution_add_column ADD COLUMNS (extra
STRING)")
- spark.sql(
- """
- INSERT INTO data_evolution_add_column VALUES
- (3, 'carol', 300, 'new'),
- (4, 'dave', 400, 'new')
- """
- )
- # MERGE INTO to trigger merge_files_by_columns with schema evolution.
- spark.sql(
- """
- CREATE TABLE IF NOT EXISTS data_evolution_add_column_updates (
- id INT,
- name STRING
- ) USING paimon
- """
- )
- spark.sql(
- "INSERT INTO data_evolution_add_column_updates VALUES (1, 'alice-v2')"
- )
- spark.sql(
- """
- MERGE INTO data_evolution_add_column t
- USING data_evolution_add_column_updates s
- ON t.id = s.id
- WHEN MATCHED THEN UPDATE SET t.name = s.name
- """
- )
- spark.sql("DROP TABLE data_evolution_add_column_updates")
-
- # ===== Data Evolution + Schema Evolution: Type Promotion =====
- # Combines data-evolution with ALTER TABLE ALTER COLUMN TYPE (INT ->
BIGINT).
- # Old files have INT; new files have BIGINT. MERGE INTO updates some rows.
- # Reader must cast old INT columns to BIGINT AND merge columns across
files.
- spark.sql(
- """
- CREATE TABLE IF NOT EXISTS data_evolution_type_promotion (
- id INT,
- value INT
- ) USING paimon
- TBLPROPERTIES (
- 'row-tracking.enabled' = 'true',
- 'data-evolution.enabled' = 'true'
- )
- """
- )
- spark.sql(
- "INSERT INTO data_evolution_type_promotion VALUES (1, 100), (2, 200)"
- )
- spark.sql(
- "ALTER TABLE data_evolution_type_promotion ALTER COLUMN value TYPE
BIGINT"
- )
- spark.sql(
- "INSERT INTO data_evolution_type_promotion VALUES (3, 3000000000)"
- )
- # MERGE INTO to trigger merge_files_by_columns with type promotion.
- spark.sql(
- """
- CREATE TABLE IF NOT EXISTS data_evolution_type_promotion_updates (
- id INT,
- value BIGINT
- ) USING paimon
- """
- )
- spark.sql(
- "INSERT INTO data_evolution_type_promotion_updates VALUES (1, 999)"
- )
- spark.sql(
- """
- MERGE INTO data_evolution_type_promotion t
- USING data_evolution_type_promotion_updates s
- ON t.id = s.id
- WHEN MATCHED THEN UPDATE SET t.value = s.value
- """
- )
- spark.sql("DROP TABLE data_evolution_type_promotion_updates")
-
- # ===== Data Evolution + Drop Column: tests NULL-fill when no file
provides a column =====
- # After MERGE INTO on old rows, the merge group files all predate ADD
COLUMN.
- # SELECT on the new column should return NULLs for old rows (not silently
drop them).
- spark.sql(
- """
- CREATE TABLE IF NOT EXISTS data_evolution_drop_column (
- id INT,
- name STRING,
- value INT
- ) USING paimon
- TBLPROPERTIES (
- 'row-tracking.enabled' = 'true',
- 'data-evolution.enabled' = 'true'
- )
- """
- )
- spark.sql(
- """
- INSERT INTO data_evolution_drop_column VALUES
- (1, 'alice', 100),
- (2, 'bob', 200)
- """
- )
- # MERGE INTO to create a partial-column file in the same row_id range.
- spark.sql(
- """
- CREATE TABLE IF NOT EXISTS data_evolution_drop_column_updates (
- id INT,
- name STRING
- ) USING paimon
- """
- )
- spark.sql(
- "INSERT INTO data_evolution_drop_column_updates VALUES (1, 'alice-v2')"
- )
- spark.sql(
- """
- MERGE INTO data_evolution_drop_column t
- USING data_evolution_drop_column_updates s
- ON t.id = s.id
- WHEN MATCHED THEN UPDATE SET t.name = s.name
- """
- )
- spark.sql("DROP TABLE data_evolution_drop_column_updates")
- # Add a new column that no existing file contains.
- spark.sql("ALTER TABLE data_evolution_drop_column ADD COLUMNS (extra
STRING)")
- # Insert new rows that DO have the extra column.
- spark.sql(
- """
- INSERT INTO data_evolution_drop_column VALUES
- (3, 'carol', 300, 'new')
- """
- )
-
- # ===== Schema Evolution: Drop Column =====
- # Old files have (id, name, score); after ALTER TABLE DROP COLUMN, table
has (id, name).
- # Reader should ignore the dropped column when reading old files.
- spark.sql(
- """
- CREATE TABLE IF NOT EXISTS schema_evolution_drop_column (
- id INT,
- name STRING,
- score INT
- ) USING paimon
- """
- )
- spark.sql(
- """
- INSERT INTO schema_evolution_drop_column VALUES
- (1, 'alice', 100),
- (2, 'bob', 200)
- """
- )
- spark.sql("ALTER TABLE schema_evolution_drop_column DROP COLUMN score")
- spark.sql(
- """
- INSERT INTO schema_evolution_drop_column VALUES
- (3, 'carol'),
- (4, 'dave')
- """
- )
-
- # ===== Schema Evolution: Rename Column across mixed file formats =====
- # Old files have physical column name 'payload'; after RENAME COLUMN, new
files
- # have 'renamed_payload'. Reader should map by field id and expose the new
name.
- spark.sql(
- """
- CREATE TABLE IF NOT EXISTS schema_evolution_rename_column (
- id INT,
- payload STRING
- ) USING paimon
- TBLPROPERTIES (
- 'file.format' = 'parquet'
- )
- """
- )
- spark.sql(
- """
- INSERT INTO schema_evolution_rename_column VALUES
- (1, 'parquet-old'),
- (2, 'parquet-old-2')
- """
- )
- spark.sql(
- "ALTER TABLE schema_evolution_rename_column RENAME COLUMN payload TO
renamed_payload"
- )
- spark.sql("ALTER TABLE schema_evolution_rename_column SET TBLPROPERTIES
('file.format' = 'orc')")
- spark.sql(
- """
- INSERT INTO schema_evolution_rename_column VALUES
- (3, 'orc-new')
- """
- )
- spark.sql("ALTER TABLE schema_evolution_rename_column SET TBLPROPERTIES
('file.format' = 'avro')")
- spark.sql(
- """
- INSERT INTO schema_evolution_rename_column VALUES
- (4, 'avro-new')
- """
- )
-
- # ===== Mixed-format Schema Evolution: Drop Column =====
- # Old Parquet/ORC files have (id, name, score); after DROP COLUMN, Avro
files
- # have only (id, name). Reader should ignore the dropped column in old
files.
- spark.sql(
- """
- CREATE TABLE IF NOT EXISTS mixed_format_schema_evolution_drop_column (
- id INT,
- name STRING,
- score INT
- ) USING paimon
- TBLPROPERTIES (
- 'file.format' = 'parquet'
- )
- """
- )
- spark.sql(
- """
- INSERT INTO mixed_format_schema_evolution_drop_column VALUES
- (1, 'parquet-alice', 100),
- (2, 'parquet-bob', 200)
- """
- )
- spark.sql(
- "ALTER TABLE mixed_format_schema_evolution_drop_column SET
TBLPROPERTIES ('file.format' = 'orc')"
- )
- spark.sql(
- """
- INSERT INTO mixed_format_schema_evolution_drop_column VALUES
- (3, 'orc-carol', 300),
- (4, 'orc-dave', 400)
- """
- )
- spark.sql("ALTER TABLE mixed_format_schema_evolution_drop_column DROP
COLUMN score")
- spark.sql(
- "ALTER TABLE mixed_format_schema_evolution_drop_column SET
TBLPROPERTIES ('file.format' = 'avro')"
- )
- spark.sql(
- """
- INSERT INTO mixed_format_schema_evolution_drop_column VALUES
- (5, 'avro-eve'),
- (6, 'avro-frank')
- """
- )
-
- # ===== Mixed-format Schema Evolution: Reorder/Move Column =====
- # Old Parquet files use the original order (id, left_value, right_value).
- # ORC and Avro files are written after moving columns; readers should
expose
- # the current table schema order and map old/new files by field id.
- spark.sql(
- """
- CREATE TABLE IF NOT EXISTS
mixed_format_schema_evolution_reorder_move_column (
- id INT,
- left_value STRING,
- right_value STRING
- ) USING paimon
- TBLPROPERTIES (
- 'file.format' = 'parquet'
- )
- """
- )
- spark.sql(
- """
- INSERT INTO mixed_format_schema_evolution_reorder_move_column VALUES
- (1, 'parquet-left-1', 'parquet-right-1'),
- (2, 'parquet-left-2', 'parquet-right-2')
- """
- )
- spark.sql(
- "ALTER TABLE mixed_format_schema_evolution_reorder_move_column ALTER
COLUMN right_value FIRST"
- )
- spark.sql(
- "ALTER TABLE mixed_format_schema_evolution_reorder_move_column SET
TBLPROPERTIES ('file.format' = 'orc')"
- )
- spark.sql(
- """
- INSERT INTO mixed_format_schema_evolution_reorder_move_column VALUES
- ('orc-right-3', 3, 'orc-left-3'),
- ('orc-right-4', 4, 'orc-left-4')
- """
- )
- spark.sql(
- "ALTER TABLE mixed_format_schema_evolution_reorder_move_column ALTER
COLUMN left_value AFTER right_value"
- )
- spark.sql(
- "ALTER TABLE mixed_format_schema_evolution_reorder_move_column SET
TBLPROPERTIES ('file.format' = 'avro')"
- )
- spark.sql(
- """
- INSERT INTO mixed_format_schema_evolution_reorder_move_column VALUES
- ('avro-right-5', 'avro-left-5', 5),
- ('avro-right-6', 'avro-left-6', 6)
- """
- )
-
- # ===== Complex Types table: ARRAY, MAP, STRUCT =====
- spark.sql(
- """
- CREATE TABLE IF NOT EXISTS complex_type_table (
- id INT,
- int_array ARRAY<INT>,
- string_map MAP<STRING, INT>,
- row_field STRUCT<name: STRING, value: INT>
- ) USING paimon
- """
- )
- spark.sql(
- """
- INSERT INTO complex_type_table VALUES
- (1, array(1, 2, 3), map('a', 10, 'b', 20), named_struct('name',
'alice', 'value', 100)),
- (2, array(4, 5), map('c', 30), named_struct('name', 'bob',
'value', 200)),
- (3, array(), map(), named_struct('name', 'carol', 'value', 300))
- """
- )
-
- # ===== Non-PK table with deletion vectors enabled =====
- # Append-only table with DV: level-0 files should NOT be filtered out
- # because there is no primary key merge.
- spark.sql(
- """
- CREATE TABLE IF NOT EXISTS simple_dv_log_table (
- id INT,
- name STRING
- ) USING paimon
- TBLPROPERTIES (
- 'deletion-vectors.enabled' = 'true'
- )
- """
- )
- spark.sql(
- """
- INSERT INTO simple_dv_log_table VALUES
- (1, 'alice'),
- (2, 'bob'),
- (3, 'carol')
- """
- )
-
- # ===== Postpone bucket PK table (bucket = -2) =====
- # New data lands in bucket-postpone and is NOT visible to readers until
compacted.
- # Without running compaction, the table should appear empty to batch
readers.
- spark.sql(
- """
- CREATE TABLE IF NOT EXISTS postpone_bucket_pk_table (
- id INT,
- name STRING
- ) USING paimon
- TBLPROPERTIES (
- 'primary-key' = 'id',
- 'bucket' = '-2',
- 'deletion-vectors.enabled' = 'true'
- )
- """
- )
- spark.sql(
- """
- INSERT INTO postpone_bucket_pk_table VALUES
- (1, 'alice'),
- (2, 'bob'),
- (3, 'carol')
- """
- )
-
- # Empty postpone table for Go fixed-bucket write tests.
- spark.sql(
- """
- CREATE TABLE IF NOT EXISTS postpone_fixed_bucket_pk_table (
- id INT,
- name STRING,
- dt STRING
- ) USING paimon
- PARTITIONED BY (dt)
- TBLPROPERTIES (
- 'primary-key' = 'id,dt',
- 'bucket' = '-2'
- )
- """
- )
-
- # ===== Dynamic bucket PK table (bucket=-1) =====
- # Two commits with overlapping keys to exercise dynamic bucket assignment
- # and index file generation. Used to verify that paimon-rust produces
- # identical hash index values when writing the same data.
- spark.sql(
- """
- CREATE TABLE IF NOT EXISTS dynamic_bucket_pk_table (
- id INT,
- name STRING
- ) USING paimon
- TBLPROPERTIES (
- 'primary-key' = 'id',
- 'bucket' = '-1'
- )
- """
- )
- spark.sql(
- """
- INSERT INTO dynamic_bucket_pk_table VALUES
- (1, 'alice'),
- (2, 'bob'),
- (3, 'carol')
- """
- )
- spark.sql(
- """
- INSERT INTO dynamic_bucket_pk_table VALUES
- (2, 'bob-v2'),
- (3, 'carol-v2'),
- (4, 'dave')
- """
- )
-
- # ===== Multi-bucket PK table for bucket predicate filtering tests =====
- # PK table with bucket=4 so data distributes across multiple buckets.
- # Bucket key defaults to primary key (id). Used to test that bucket
predicate
- # filtering correctly computes target buckets from equality predicates.
- spark.sql(
- """
- CREATE TABLE IF NOT EXISTS multi_bucket_pk_table (
- id INT,
- name STRING
- ) USING paimon
- TBLPROPERTIES (
- 'primary-key' = 'id',
- 'bucket' = '4',
- 'deletion-vectors.enabled' = 'true'
- )
- """
- )
- spark.sql(
- """
- INSERT INTO multi_bucket_pk_table VALUES
- (1, 'alice'),
- (2, 'bob'),
- (3, 'carol'),
- (4, 'dave'),
- (5, 'eve'),
- (6, 'frank'),
- (7, 'grace'),
- (8, 'heidi')
- """
- )
-
- # ===== String bucket key tables for variable-length hash tests =====
- # Short string keys (<=7 bytes) use inline encoding in BinaryRow.
- spark.sql(
- """
- CREATE TABLE IF NOT EXISTS string_bucket_short_key (
- code STRING,
- value INT
- ) USING paimon
- TBLPROPERTIES (
- 'primary-key' = 'code',
- 'bucket' = '4',
- 'deletion-vectors.enabled' = 'true'
- )
- """
- )
- spark.sql(
- """
- INSERT INTO string_bucket_short_key VALUES
- ('aaa', 1),
- ('bbb', 2),
- ('ccc', 3),
- ('ddd', 4),
- ('eee', 5),
- ('fff', 6),
- ('ggg', 7),
- ('hhh', 8)
- """
- )
-
- # Long string keys (>7 bytes) use variable-length encoding in BinaryRow.
- spark.sql(
- """
- CREATE TABLE IF NOT EXISTS string_bucket_long_key (
- code STRING,
- value INT
- ) USING paimon
- TBLPROPERTIES (
- 'primary-key' = 'code',
- 'bucket' = '4',
- 'deletion-vectors.enabled' = 'true'
- )
- """
- )
- spark.sql(
- """
- INSERT INTO string_bucket_long_key VALUES
- ('alpha-long-key', 1),
- ('bravo-long-key', 2),
- ('charlie-long-key', 3),
- ('delta-long-key', 4),
- ('echo-long-key', 5),
- ('foxtrot-long-key', 6),
- ('golf-long-key', 7),
- ('hotel-long-key', 8)
- """
- )
-
- # ===== Full types table: parquet, orc, avro =====
- # Note: Spark 3.x does not support parameterized timestamp precision (e.g.
TIMESTAMP(3)),
- # so all timestamps here use the default precision 6 (microseconds).
- spark.sql(
- """
- CREATE TABLE IF NOT EXISTS full_types_table (
- id INT,
- col_boolean BOOLEAN,
- col_tinyint TINYINT,
- col_smallint SMALLINT,
- col_int INT,
- col_bigint BIGINT,
- col_float FLOAT,
- col_double DOUBLE,
- col_decimal DECIMAL(10, 2),
- col_decimal5 DECIMAL(5),
- col_decimal38 DECIMAL(38, 18),
- col_string STRING,
- col_binary BINARY,
- col_date DATE,
- col_timestamp TIMESTAMP_NTZ,
- col_timestamp_ltz TIMESTAMP,
- col_array ARRAY<INT>,
- col_map MAP<STRING, INT>,
- col_struct STRUCT<name: STRING, value: INT>
- ) USING paimon
- TBLPROPERTIES (
- 'file.format' = 'parquet'
- )
- """
- )
- # Parquet row
- spark.sql(
- """
- INSERT INTO full_types_table VALUES
- (1, true, CAST(1 AS TINYINT), CAST(100 AS SMALLINT), 1000, 100000,
- CAST(1.5 AS FLOAT), 2.5, CAST(123.45 AS DECIMAL(10,2)),
- CAST(12345 AS DECIMAL(5)), CAST(12345.678901234567890 AS
DECIMAL(38,18)),
- 'parquet-hello', X'DEADBEEF',
- DATE '2024-01-01',
- TIMESTAMP_NTZ '2024-01-01 10:00:00.123456',
- TIMESTAMP '2024-01-01 10:00:00.123456',
- array(1, 2, 3), map('a', 10, 'b', 20),
- named_struct('name', 'alice', 'value', 100))
- """
- )
- # Switch to ORC
- spark.sql("ALTER TABLE full_types_table SET TBLPROPERTIES ('file.format' =
'orc')")
- spark.sql(
- """
- INSERT INTO full_types_table VALUES
- (2, false, CAST(2 AS TINYINT), CAST(200 AS SMALLINT), 2000, 200000,
- CAST(3.5 AS FLOAT), 4.5, CAST(678.90 AS DECIMAL(10,2)),
- CAST(99999 AS DECIMAL(5)), CAST(99999.999999999999999 AS
DECIMAL(38,18)),
- 'orc-world', X'CAFEBABE',
- DATE '2024-06-15',
- TIMESTAMP_NTZ '2024-06-15 12:30:00.456789',
- TIMESTAMP '2024-06-15 12:30:00.456789',
- array(4, 5), map('c', 30),
- named_struct('name', 'bob', 'value', 200))
- """
- )
- # Switch to Avro
- spark.sql("ALTER TABLE full_types_table SET TBLPROPERTIES ('file.format' =
'avro')")
- spark.sql(
- """
- INSERT INTO full_types_table VALUES
- (3, true, CAST(3 AS TINYINT), CAST(300 AS SMALLINT), 3000, 300000,
- CAST(5.5 AS FLOAT), 6.5, CAST(999.99 AS DECIMAL(10,2)),
- CAST(0 AS DECIMAL(5)), CAST(0.000000000000000001 AS
DECIMAL(38,18)),
- 'avro-test', X'01020304',
- DATE '2025-12-31',
- TIMESTAMP_NTZ '2025-12-31 23:59:59.999999',
- TIMESTAMP '2025-12-31 23:59:59.999999',
- array(6), map('d', 40, 'e', 50),
- named_struct('name', 'carol', 'value', 300))
- """
- )
-
- # ===== Full types boundary table: parquet, orc, avro =====
- # Each format writes one boundary row and one all-null row for nullable
fields.
- spark.sql(
- """
- CREATE TABLE IF NOT EXISTS full_types_boundary_table (
- id INT,
- col_boolean BOOLEAN,
- col_tinyint TINYINT,
- col_smallint SMALLINT,
- col_int INT,
- col_bigint BIGINT,
- col_float FLOAT,
- col_double DOUBLE,
- col_decimal DECIMAL(10, 2),
- col_decimal5 DECIMAL(5),
- col_decimal38 DECIMAL(38, 18),
- col_string STRING,
- col_binary BINARY,
- col_date DATE,
- col_timestamp TIMESTAMP_NTZ,
- col_timestamp_ltz TIMESTAMP,
- col_array ARRAY<INT>,
- col_map MAP<STRING, INT>,
- col_struct STRUCT<name: STRING, value: INT>
- ) USING paimon
- TBLPROPERTIES (
- 'file.format' = 'parquet'
- )
- """
- )
- spark.sql(
- """
- INSERT INTO full_types_boundary_table VALUES
- (1, false, CAST('-128' AS TINYINT), CAST('-32768' AS SMALLINT),
- CAST('-2147483648' AS INT), CAST('-9223372036854775808' AS
BIGINT),
- CAST(-0.5 AS FLOAT), -1.25, CAST('-99999999.99' AS DECIMAL(10,2)),
- CAST('-99999' AS DECIMAL(5)),
- CAST('-99999999999999999999.999999999999999999' AS
DECIMAL(38,18)),
- '', X'',
- DATE '1969-12-31',
- TIMESTAMP_NTZ '1970-01-01 00:00:00.000001',
- TIMESTAMP '1970-01-01 00:00:00.000001',
- array(CAST(NULL AS INT), CAST('-2147483648' AS INT), CAST(0 AS
INT)),
- map('negative', CAST('-2147483648' AS INT), 'zero', CAST(NULL AS
INT)),
- named_struct('name', CAST(NULL AS STRING), 'value', CAST(-1 AS
INT))),
- (2, CAST(NULL AS BOOLEAN), CAST(NULL AS TINYINT), CAST(NULL AS
SMALLINT),
- CAST(NULL AS INT), CAST(NULL AS BIGINT),
- CAST(NULL AS FLOAT), CAST(NULL AS DOUBLE), CAST(NULL AS
DECIMAL(10,2)),
- CAST(NULL AS DECIMAL(5)), CAST(NULL AS DECIMAL(38,18)),
- CAST(NULL AS STRING), CAST(NULL AS BINARY),
- CAST(NULL AS DATE),
- CAST(NULL AS TIMESTAMP_NTZ),
- CAST(NULL AS TIMESTAMP),
- CAST(NULL AS ARRAY<INT>),
- CAST(NULL AS MAP<STRING, INT>),
- CAST(NULL AS STRUCT<name: STRING, value: INT>))
- """
- )
- spark.sql("ALTER TABLE full_types_boundary_table SET TBLPROPERTIES
('file.format' = 'orc')")
- spark.sql(
- """
- INSERT INTO full_types_boundary_table VALUES
- (3, true, CAST('127' AS TINYINT), CAST('32767' AS SMALLINT),
- CAST('2147483647' AS INT), CAST('9223372036854775807' AS BIGINT),
- CAST(0.25 AS FLOAT), 0.5, CAST('99999999.99' AS DECIMAL(10,2)),
- CAST('99999' AS DECIMAL(5)),
- CAST('99999999999999999999.999999999999999999' AS DECIMAL(38,18)),
- 'orc-boundary', X'00FF',
- DATE '1970-01-01',
- TIMESTAMP_NTZ '1970-01-01 00:00:00',
- TIMESTAMP '1970-01-01 00:00:00',
- CAST(array() AS ARRAY<INT>),
- CAST(map() AS MAP<STRING, INT>),
- named_struct('name', 'orc', 'value', CAST(NULL AS INT))),
- (4, CAST(NULL AS BOOLEAN), CAST(NULL AS TINYINT), CAST(NULL AS
SMALLINT),
- CAST(NULL AS INT), CAST(NULL AS BIGINT),
- CAST(NULL AS FLOAT), CAST(NULL AS DOUBLE), CAST(NULL AS
DECIMAL(10,2)),
- CAST(NULL AS DECIMAL(5)), CAST(NULL AS DECIMAL(38,18)),
- CAST(NULL AS STRING), CAST(NULL AS BINARY),
- CAST(NULL AS DATE),
- CAST(NULL AS TIMESTAMP_NTZ),
- CAST(NULL AS TIMESTAMP),
- CAST(NULL AS ARRAY<INT>),
- CAST(NULL AS MAP<STRING, INT>),
- CAST(NULL AS STRUCT<name: STRING, value: INT>))
- """
- )
- spark.sql("ALTER TABLE full_types_boundary_table SET TBLPROPERTIES
('file.format' = 'avro')")
- spark.sql(
- """
- INSERT INTO full_types_boundary_table VALUES
- (5, false, CAST(0 AS TINYINT), CAST(0 AS SMALLINT),
- 0, 0,
- CAST(0.0 AS FLOAT), 0.0, CAST(0.00 AS DECIMAL(10,2)),
- CAST(0 AS DECIMAL(5)), CAST(0 AS DECIMAL(38,18)),
- 'avro-boundary', X'0102',
- DATE '1970-01-02',
- TIMESTAMP_NTZ '1970-01-01 00:00:00.999999',
- TIMESTAMP '1970-01-01 00:00:00.999999',
- array(CAST(7 AS INT)),
- map('seven', 7),
- named_struct('name', 'avro', 'value', 7)),
- (6, CAST(NULL AS BOOLEAN), CAST(NULL AS TINYINT), CAST(NULL AS
SMALLINT),
- CAST(NULL AS INT), CAST(NULL AS BIGINT),
- CAST(NULL AS FLOAT), CAST(NULL AS DOUBLE), CAST(NULL AS
DECIMAL(10,2)),
- CAST(NULL AS DECIMAL(5)), CAST(NULL AS DECIMAL(38,18)),
- CAST(NULL AS STRING), CAST(NULL AS BINARY),
- CAST(NULL AS DATE),
- CAST(NULL AS TIMESTAMP_NTZ),
- CAST(NULL AS TIMESTAMP),
- CAST(NULL AS ARRAY<INT>),
- CAST(NULL AS MAP<STRING, INT>),
- CAST(NULL AS STRUCT<name: STRING, value: INT>))
- """
- )
-
-
- # ===== First-Row merge engine PK table =====
- # first-row keeps the earliest inserted row per key; later duplicates are
ignored.
- # After compaction, level-0 files are promoted so the batch reader (which
skips
- # level-0 for first-row) can see the compacted data.
- spark.sql(
- """
- CREATE TABLE IF NOT EXISTS first_row_pk_table (
- id INT,
- name STRING
- ) USING paimon
- TBLPROPERTIES (
- 'primary-key' = 'id',
- 'bucket' = '1',
- 'merge-engine' = 'first-row'
- )
- """
- )
- spark.sql(
- """
- INSERT INTO first_row_pk_table VALUES
- (1, 'alice'),
- (2, 'bob'),
- (3, 'carol')
- """
- )
- spark.sql(
- """
- INSERT INTO first_row_pk_table VALUES
- (2, 'bob-v2'),
- (3, 'carol-v2'),
- (4, 'dave')
- """
- )
- # Compact to promote level-0 files so the batch reader can see them.
- spark.sql("CALL sys.compact('default.first_row_pk_table')")
-
-
-if __name__ == "__main__":
- main()
diff --git a/dev/spark/spark-defaults.conf b/dev/spark/spark-defaults.conf
deleted file mode 100644
index 8f314858..00000000
--- a/dev/spark/spark-defaults.conf
+++ /dev/null
@@ -1,25 +0,0 @@
-#
-# 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.
-#
-
-# Paimon filesystem catalog – warehouse is mounted at /tmp/paimon-warehouse
-spark.sql.catalog.paimon
org.apache.paimon.spark.SparkCatalog
-spark.sql.catalog.paimon.warehouse file:/tmp/paimon-warehouse
-spark.sql.extensions
org.apache.paimon.spark.extensions.PaimonSparkSessionExtensions
-
-spark.sql.defaultCatalog paimon
diff --git a/docs/src/contributing.md b/docs/src/contributing.md
index 6fb2e4ca..c0a057a5 100644
--- a/docs/src/contributing.md
+++ b/docs/src/contributing.md
@@ -51,16 +51,17 @@ cargo fmt --all
cargo clippy --locked --all-targets --workspace --features fulltext,vortex --
-D warnings
```
-The integration suites read a Paimon warehouse written by Spark, so they need
-Docker rather than `cargo` alone:
+The integration suites automatically extract a committed Java-written golden
+warehouse into temporary directories; Spark and Docker are not required:
```bash
-make docker-up # writes /tmp/paimon-warehouse
cargo test -p paimon-integration-tests --all-targets
cargo test -p paimon-datafusion --all-targets
-make docker-down
```
+Set `PAIMON_TEST_WAREHOUSE` to test an existing warehouse instead. Missing
fixture
+tables fail the tests rather than being skipped or generated by the Rust
writer.
+
## Finding Issues
- Check [open issues](https://github.com/apache/paimon-rust/issues) for tasks
to work on
diff --git a/docs/src/getting-started.md b/docs/src/getting-started.md
index 26cb18f8..bd19817b 100644
--- a/docs/src/getting-started.md
+++ b/docs/src/getting-started.md
@@ -553,8 +553,6 @@ cargo build
# Unit tests
cargo test
-# Integration tests (requires Docker)
-make docker-up
+# Integration tests (automatically extract the committed golden warehouse)
cargo test -p paimon-integration-tests
-make docker-down
```
