Added logic to create table from schema. (#219)

Author: subkanthi

ice-rest-catalog/src/test/java/com/altinity/ice/rest/catalog/DockerElasticMQWatchIT.java

@@ -80,7 +80,7 @@ public class DockerElasticMQWatchIT {
   private static final String DEFAULT_CATALOG_IMAGE =
       "altinity/ice-rest-catalog:debug-with-ice-local";
   private static final String DEFAULT_ELASTICMQ_IMAGE = "softwaremill/elasticmq-native:1.6.15";
-  private static final String DEFAULT_MINIO_IMAGE = "quay.io/minio/minio:latest";
+  private static final String DEFAULT_MINIO_IMAGE = "rustfs/rustfs:1.0.0";
 
   private static final String BUCKET = "test-bucket";
   private static final String QUEUE_NAME = "s3-events";
@@ -116,10 +116,10 @@ public void setUp() throws Exception {
             .withNetwork(network)
             .withNetworkAliases("minio")
             .withExposedPorts(9000)
-            .withEnv("MINIO_ACCESS_KEY", "minioadmin")
-            .withEnv("MINIO_SECRET_KEY", "minioadmin")
-            .withCommand("server", "/data")
-            .waitingFor(Wait.forHttp("/minio/health/live").forPort(9000));
+            .withEnv("RUSTFS_ACCESS_KEY", "minioadmin")
+            .withEnv("RUSTFS_SECRET_KEY", "minioadmin")
+            .withCommand("/data")
+            .waitingFor(Wait.forHttp("/health").forPort(9000));
     minio.start();
 
     String minioHostEndpoint = "http://" + minio.getHost() + ":" + minio.getMappedPort(9000);

ice-rest-catalog/src/test/java/com/altinity/ice/rest/catalog/DockerScenarioBasedIT.java

@@ -20,6 +20,7 @@
 import org.testcontainers.containers.GenericContainer;
 import org.testcontainers.containers.Network;
 import org.testcontainers.containers.wait.strategy.Wait;
+import org.testcontainers.images.builder.ImageFromDockerfile;
 import org.testcontainers.utility.MountableFile;
 import org.testng.annotations.AfterClass;
 import org.testng.annotations.BeforeClass;
@@ -49,6 +50,8 @@ public class DockerScenarioBasedIT extends RESTCatalogTestBase {
 
   private GenericContainer<?> clickhouse;
 
+  private GenericContainer<?> pyiceberg;
+
   @Override
   @BeforeClass
   @SuppressWarnings("resource")
@@ -61,14 +64,14 @@ public void setUp() throws Exception {
 
     // Start MinIO
     minio =
-        new GenericContainer<>("quay.io/minio/minio:latest")
+        new GenericContainer<>("rustfs/rustfs:1.0.0")
             .withNetwork(network)
             .withNetworkAliases("minio")
             .withExposedPorts(9000)
-            .withEnv("MINIO_ACCESS_KEY", "minioadmin")
-            .withEnv("MINIO_SECRET_KEY", "minioadmin")
-            .withCommand("server", "/data")
-            .waitingFor(Wait.forHttp("/minio/health/live").forPort(9000));
+            .withEnv("RUSTFS_ACCESS_KEY", "minioadmin")
+            .withEnv("RUSTFS_SECRET_KEY", "minioadmin")
+            .withCommand("/data")
+            .waitingFor(Wait.forHttp("/health").forPort(9000));
     minio.start();
 
     // Create test bucket via MinIO's host-mapped port
@@ -149,6 +152,29 @@ public void setUp() throws Exception {
       throw e;
     }
 
+    // PyIceberg container: independent Iceberg implementation used by scenarios (validate.py)
+    // to cross-check tables/data written by ice. Built locally from test resources.
+    URL pyicebergDockerfile = getClass().getClassLoader().getResource("pyiceberg/Dockerfile");
+    if (pyicebergDockerfile == null) {
+      throw new IllegalStateException("pyiceberg/Dockerfile not found on classpath");
+    }
+    pyiceberg =
+        new GenericContainer<>(
+                new ImageFromDockerfile("ice-test-pyiceberg", false)
+                    .withDockerfile(Paths.get(pyicebergDockerfile.toURI())))
+            .withNetwork(network)
+            .withNetworkAliases("pyiceberg")
+            .withCopyFileToContainer(MountableFile.forHostPath(scenariosDir), "/scenarios")
+            .withCommand("sleep infinity");
+    try {
+      pyiceberg.start();
+    } catch (Exception e) {
+      if (pyiceberg != null) {
+        logger.error("PyIceberg container logs: {}", pyiceberg.getLogs());
+      }
+      throw e;
+    }
+
     // Copy CLI config into container so ice CLI can talk to co-located REST server.
     // The s3 section lets CLI commands that read s3:// paths directly (e.g. describe-metadata on
     // metadata.json) reach MinIO via its network alias instead of defaulting to AWS.
@@ -180,6 +206,9 @@ public void setUp() throws Exception {
   @Override
   @AfterClass
   public void tearDown() {
+    if (pyiceberg != null) {
+      pyiceberg.close();
+    }
     if (clickhouse != null) {
       clickhouse.close();
     }
@@ -220,9 +249,25 @@ protected ScenarioTestRunner createScenarioRunner(String scenarioName) throws Ex
           "Could not set ClickHouse wrapper script executable: " + chWrapperScript);
     }
 
+    // Wrapper script on host: docker exec <pyiceberg container> python3 "$@" (runs validate.py
+    // inside the pyiceberg container; /scenarios is mounted at the same path as in catalog)
+    File pyWrapperScript = File.createTempFile("py-docker-exec-", ".sh");
+    pyWrapperScript.deleteOnExit();
+    Files.writeString(
+        pyWrapperScript.toPath(),
+        "#!/bin/sh\nexec docker exec -e CATALOG_URI_INTERNAL=http://catalog:5000"
+            + " -e S3_ENDPOINT_INTERNAL=http://minio:9000 "
+            + pyiceberg.getContainerId()
+            + " python3 \"$@\"\n");
+    if (!pyWrapperScript.setExecutable(true)) {
+      throw new IllegalStateException(
+          "Could not set PyIceberg wrapper script executable: " + pyWrapperScript);
+    }
+
     Map<String, String> templateVars = new HashMap<>();
     templateVars.put("ICE_CLI", wrapperScript.getAbsolutePath());
     templateVars.put("CH_EXEC", chWrapperScript.getAbsolutePath());
+    templateVars.put("PY_EXEC", pyWrapperScript.getAbsolutePath());
     templateVars.put("CLI_CONFIG", "/tmp/ice-cli.yaml");
     templateVars.put("SCENARIO_DIR", "/scenarios/" + scenarioName);
     templateVars.put("MINIO_ENDPOINT", "");

ice-rest-catalog/src/test/java/com/altinity/ice/rest/catalog/RESTCatalogTestBase.java

@@ -50,11 +50,11 @@ public abstract class RESTCatalogTestBase {
 
   @SuppressWarnings("rawtypes")
   protected final GenericContainer minio =
-      new GenericContainer("quay.io/minio/minio:latest")
+      new GenericContainer("rustfs/rustfs:1.0.0")
           .withExposedPorts(9000)
-          .withEnv("MINIO_ACCESS_KEY", "minioadmin")
-          .withEnv("MINIO_SECRET_KEY", "minioadmin")
-          .withCommand("server", "/data");
+          .withEnv("RUSTFS_ACCESS_KEY", "minioadmin")
+          .withEnv("RUSTFS_SECRET_KEY", "minioadmin")
+          .withCommand("/data");
 
   @BeforeClass
   public void setUp() throws Exception {

ice-rest-catalog/src/test/java/com/altinity/ice/rest/catalog/ScenarioBasedIT.java

@@ -39,6 +39,8 @@ protected ScenarioTestRunner createScenarioRunner(String scenarioName) throws Ex
     templateVars.put("CATALOG_URI", getCatalogUri());
     // DockerScenarioBasedIT sets these for ClickHouse; empty so basic-operations skips CH block.
     templateVars.put("CH_EXEC", "");
+    // DockerScenarioBasedIT sets this for PyIceberg; empty so scenarios skip pyiceberg validation.
+    templateVars.put("PY_EXEC", "");
     templateVars.put("CATALOG_URI_INTERNAL", "");
     templateVars.put("S3_ENDPOINT_INTERNAL", "");
 

ice-rest-catalog/src/test/pyiceberg/uv.lock

@@ -0,0 +1,5 @@
+# PyIceberg test image for scenario ITs: independent Iceberg reader used to validate
+# tables/data written by ice. Built at test time via Testcontainers ImageFromDockerfile.
+FROM python:3.12-slim
+ARG PYICEBERG_VERSION=0.12.0
+RUN pip install --no-cache-dir "pyiceberg[pyarrow,s3fs]==${PYICEBERG_VERSION}"

ice-rest-catalog/src/test/resources/pyiceberg/Dockerfile

@@ -0,0 +1,62 @@
+#!/bin/bash
+set -e
+
+echo "Running Iceberg v3 unknown-type (--schema) test..."
+
+# Create namespace
+{{ICE_CLI}} --config {{CLI_CONFIG}} create-namespace ${NAMESPACE_NAME}
+echo "OK Created namespace: ${NAMESPACE_NAME}"
+
+# Create a v3 table with an explicit schema containing a v3-only unknown column
+# (not representable in Parquet, so --schema-from-parquet cannot express it)
+{{ICE_CLI}} --config {{CLI_CONFIG}} create-table ${TABLE_NAME} --format-version=3 \
+  --schema '[{"name":"id","type":"long","required":true},{"name":"payload","type":"unknown"}]'
+echo "OK Created table ${TABLE_NAME} via --schema (unknown column)"
+
+# Verify schema output shows the unknown column
+{{ICE_CLI}} --config {{CLI_CONFIG}} describe -s ${TABLE_NAME} > /tmp/v3_unknown_schema.txt
+if ! grep -q "payload: optional unknown" /tmp/v3_unknown_schema.txt; then
+  echo "FAIL describe -s output missing 'payload: optional unknown'"
+  cat /tmp/v3_unknown_schema.txt
+  exit 1
+fi
+echo "OK describe -s shows payload: optional unknown"
+
+# Verify describe output shows formatVersion 3 (Iceberg strips format-version from properties)
+{{ICE_CLI}} --config {{CLI_CONFIG}} describe ${TABLE_NAME} > /tmp/v3_unknown_describe.txt
+if ! grep -q "formatVersion: 3" /tmp/v3_unknown_describe.txt; then
+  echo "FAIL describe output missing 'formatVersion: 3'"
+  cat /tmp/v3_unknown_describe.txt
+  exit 1
+fi
+echo "OK describe shows formatVersion: 3"
+
+# Negative: unknown type must be rejected without --format-version=3 (default is v2)
+if {{ICE_CLI}} --config {{CLI_CONFIG}} create-table ${NAMESPACE_NAME}.t2 \
+    --schema '[{"name":"payload","type":"unknown"}]' > /tmp/v3_unknown_v2.out 2>&1; then
+  echo "FAIL create-table with unknown type on v2 succeeded unexpectedly"
+  exit 1
+fi
+if ! grep -q "not supported until v3" /tmp/v3_unknown_v2.out; then
+  echo "FAIL v2 rejection missing 'not supported until v3'"
+  cat /tmp/v3_unknown_v2.out
+  exit 1
+fi
+echo "OK v2 table with unknown type rejected"
+
+# Independent validation via pyiceberg (Docker runner only; empty PY_EXEC = skipped)
+if [ -n "{{PY_EXEC}}" ]; then
+  {{PY_EXEC}} "${SCENARIO_DIR}/validate.py" ${TABLE_NAME}
+  echo "OK pyiceberg validation passed"
+else
+  echo "SKIP pyiceberg validation (PY_EXEC not available)"
+fi
+
+# Cleanup
+{{ICE_CLI}} --config {{CLI_CONFIG}} delete-table ${TABLE_NAME}
+echo "OK Deleted table: ${TABLE_NAME}"
+
+{{ICE_CLI}} --config {{CLI_CONFIG}} delete-namespace ${NAMESPACE_NAME}
+echo "OK Deleted namespace: ${NAMESPACE_NAME}"
+
+echo "Iceberg v3 unknown-type test completed successfully"

ice-rest-catalog/src/test/resources/scenarios/iceberg-v3-unknown-type/run.sh.tmpl

@@ -0,0 +1,9 @@
+name: "Iceberg v3 Unknown Type via --schema"
+description: "Tests create-table --schema with a v3-only unknown column; verifies via ice describe and pyiceberg (independent reader)"
+
+catalogConfig:
+  warehouse: "s3://test-bucket/warehouse"
+
+env:
+  NAMESPACE_NAME: "test_v3_unknown"
+  TABLE_NAME: "test_v3_unknown.t1"

ice-rest-catalog/src/test/resources/scenarios/iceberg-v3-unknown-type/scenario.yaml

@@ -0,0 +1,56 @@
+"""Independent validation of an ice-created v3 table using PyIceberg.
+
+Runs inside the pyiceberg test container (see scenarios/pyiceberg/Dockerfile).
+Reads the table via the REST catalog and asserts:
+  - format_version == 3
+  - the payload column has the v3-only UnknownType
+  - the table scans to zero rows (unknown columns are always null per spec)
+
+Usage: python3 validate.py <table-id>   (e.g. test_v3_unknown.t1)
+"""
+
+import os
+import sys
+
+from pyiceberg.catalog import load_catalog
+from pyiceberg.types import UnknownType
+
+
+def main() -> int:
+    table_id = sys.argv[1] if len(sys.argv) > 1 else "test_v3_unknown.t1"
+    catalog_uri = os.environ.get("CATALOG_URI_INTERNAL", "http://catalog:5000")
+    s3_endpoint = os.environ.get("S3_ENDPOINT_INTERNAL", "http://minio:9000")
+
+    catalog = load_catalog(
+        "rest",
+        uri=catalog_uri,
+        warehouse="s3://test-bucket/warehouse",
+        **{
+            # ice-rest-catalog test config uses anonymous access; without this, pyiceberg
+            # sends "Authorization: Bearer None" (LegacyOAuth2AuthManager) -> 403 Invalid token
+            "auth": {"type": "noop"},
+            "s3.endpoint": s3_endpoint,
+            "s3.access-key-id": "minioadmin",
+            "s3.secret-access-key": "minioadmin",
+        },
+    )
+    table = catalog.load_table(table_id)
+
+    format_version = table.metadata.format_version
+    assert format_version == 3, f"expected format_version 3, got {format_version}"
+    print(f"OK pyiceberg: format_version == {format_version}")
+
+    field = table.schema().find_field("payload")
+    assert isinstance(
+        field.field_type, UnknownType
+    ), f"expected UnknownType for payload, got {field.field_type}"
+    print(f"OK pyiceberg: payload type is {field.field_type}")
+
+    rows = table.scan().to_arrow()
+    assert rows.num_rows == 0, f"expected 0 rows, got {rows.num_rows}"
+    print("OK pyiceberg: scan returned 0 rows")
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

ice-rest-catalog/src/test/resources/scenarios/iceberg-v3-unknown-type/validate.py

@@ -34,6 +34,14 @@ fi
 
 echo "OK Scan output contains expected columns"
 
+# Independent validation via pyiceberg (Docker runner only; empty PY_EXEC = skipped)
+if [ -n "{{PY_EXEC}}" ]; then
+  {{PY_EXEC}} "${SCENARIO_DIR}/validate.py" ${TABLE_NAME} "${INPUT_PATH}"
+  echo "OK pyiceberg validation passed"
+else
+  echo "SKIP pyiceberg validation (PY_EXEC not available)"
+fi
+
 # Cleanup
 {{ICE_CLI}} --config {{CLI_CONFIG}} delete-table ${TABLE_NAME}
 echo "OK Deleted table: ${TABLE_NAME}"