manual sync

Author: Dmitriy Fingerman

iceberg/iceberg-handler/src/test/queries/positive/iceberg_rest_catalog_gravitino.q

@@ -66,6 +66,11 @@ show create table ice_orc2;
 insert into ice_orc2 partition (company_id=100) 
 VALUES ('fn1','ln1', 1, 10), ('fn2','ln2', 2, 20), ('fn3','ln3', 3, 30);
 
+--! In CI, Testcontainers' .withFileSystemBind() is not able to bind the same host path to the same container path,
+--! so as a workaround, the .metadata.json files from container are manually synced in a daemon process,
+--! since the sync can take some time, need to wait for it to happen after the insert operation.
+! sleep 20;
+
 describe formatted ice_orc2;
 select * from ice_orc2;
 

itests/qtest-iceberg/src/test/java/org/apache/hadoop/hive/cli/TestIcebergRESTCatalogGravitinoLlapLocalCliDriver.java

@@ -18,6 +18,8 @@
 
 package org.apache.hadoop.hive.cli;
 
+import org.apache.commons.compress.archivers.tar.TarArchiveEntry;
+import org.apache.commons.compress.archivers.tar.TarArchiveInputStream;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.hive.cli.control.CliAdapter;
 import org.apache.hadoop.hive.cli.control.CliConfigs;
@@ -35,8 +37,8 @@
 import org.junit.runner.RunWith;
 import org.junit.runners.Parameterized;
 import org.junit.runners.Parameterized.Parameters;
+import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
-import org.testcontainers.containers.BindMode;
 import org.testcontainers.containers.output.Slf4jLogConsumer;
 import org.testcontainers.containers.wait.strategy.Wait;
 import org.testcontainers.containers.wait.strategy.WaitAllStrategy;
@@ -54,14 +56,18 @@
 import java.time.Duration;
 import java.util.Comparator;
 import java.util.List;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
 
 @RunWith(Parameterized.class)
 public class TestIcebergRESTCatalogGravitinoLlapLocalCliDriver {
 
+  private static final Logger LOG = LoggerFactory.getLogger(TestIcebergRESTCatalogGravitinoLlapLocalCliDriver.class);
+  
   private static final String CATALOG_NAME = "ice01";
   private static final String GRAVITINO_CONF_FILE_TEMPLATE = "gravitino-h2-test-template.conf";
   private static final String GRAVITINO_ROOT_DIR = "/root/gravitino-iceberg-rest-server";
-  private static final long GRAVITINO_STARTUP_TIMEOUT_MINUTES = 5l;
+  private static final long GRAVITINO_STARTUP_TIMEOUT_MINUTES = 5L;
 
   private static final String GRAVITINO_STARTUP_SCRIPT = GRAVITINO_ROOT_DIR + "/bin/start-iceberg-rest-server.sh";
   private static final String GRAVITINO_H2_LIB = GRAVITINO_ROOT_DIR + "/libs/h2-driver.jar";
@@ -76,6 +82,7 @@ public class TestIcebergRESTCatalogGravitinoLlapLocalCliDriver {
 
   private GenericContainer<?> gravitinoContainer;
   private Path warehouseDir;
+  private final ExecutorService fileSyncExecutor = Executors.newSingleThreadExecutor();
 
   @Parameters(name = "{0}")
   public static List<Object[]> getParameters() throws Exception {
@@ -97,7 +104,8 @@ public TestIcebergRESTCatalogGravitinoLlapLocalCliDriver(String name, File qfile
   public void setup() throws IOException {
     createWarehouseDir();
     prepareGravitinoConfig();
-    setupGravitinoContainer();
+    startGravitinoContainer();
+    startWarehouseDirSync();
 
     String host = gravitinoContainer.getHost();
     Integer port = gravitinoContainer.getMappedPort(9001);
@@ -117,6 +125,8 @@ public void teardown() {
       gravitinoContainer.stop();
     }
 
+    fileSyncExecutor.shutdownNow();
+
     if (warehouseDir != null && Files.exists(warehouseDir)) {
       try (var paths = Files.walk(warehouseDir)) {
         paths.sorted(Comparator.reverseOrder())
@@ -128,23 +138,17 @@ public void teardown() {
     }
   }
 
-  private void setupGravitinoContainer() {
+  private void startGravitinoContainer() {
     gravitinoContainer = new GenericContainer<>(GRAVITINO_IMAGE)
         .withExposedPorts(9001)
         // Update entrypoint to create the warehouse directory before starting the server
-        .withCreateContainerCmdModifier(cmd -> {
-          cmd.withEntrypoint("bash", "-c",
-              String.format("mkdir -p %s && exec %s", warehouseDir.toString(), GRAVITINO_STARTUP_SCRIPT));
-        })
+        .withCreateContainerCmdModifier(cmd -> cmd.withEntrypoint("bash", "-c",
+            String.format("mkdir -p %s && exec %s", warehouseDir.toString(), GRAVITINO_STARTUP_SCRIPT)))
         // Mount gravitino configuration file
         .withCopyFileToContainer(
             MountableFile.forHostPath(Paths.get(warehouseDir.toString(), GRAVITINO_CONF_FILE_TEMPLATE)),
             GRAVITINO_CONF_FILE
         )
-        // Bind the warehouse directory into the container on the same path to keep gravitino and hive aligned
-        .withFileSystemBind(
-            warehouseDir.toString(), warehouseDir.toString(), BindMode.READ_WRITE
-        )
         // Mount the H2 driver JAR into the server's lib directory
         .withCopyFileToContainer(
             MountableFile.forHostPath(
@@ -165,6 +169,85 @@ private void setupGravitinoContainer() {
     gravitinoContainer.start();
   }
 
+  /**
+   * Starts a background daemon that continuously synchronizes the Iceberg warehouse
+   * directory from the running Gravitino container to the host file system.
+   *
+   * <p>In CI environments, Testcontainers' {@code .withFileSystemBind()} cannot reliably
+   * bind the same host path to the same path inside the container, especially when
+   * using remote Docker hosts or Docker-in-Docker setups. This causes the container's
+   * writes (e.g., Iceberg metadata files like {@code .metadata.json}) to be invisible
+   * on the host.</p>
+   *
+   * <p>This method works around that limitation by repeatedly copying new files from
+   * the container's warehouse directory to the corresponding host directory. Existing
+   * files on the host are preserved, and only files that do not yet exist are copied.
+   * The sync runs every 1 second while the container is running.</p>
+   *
+   * <p>Each archive copy from the container is extracted using a {@link TarArchiveInputStream},
+   * and directories are created as needed. Files that already exist on the host are skipped
+   * to avoid overwriting container data.</p>
+   */
+  private void startWarehouseDirSync() {
+    fileSyncExecutor.submit(() -> {
+      try {
+        while (gravitinoContainer.isRunning()) {
+          try (InputStream tarStream = gravitinoContainer.getDockerClient()
+              .copyArchiveFromContainerCmd(gravitinoContainer.getContainerId(), warehouseDir.toString())
+              .exec();
+               TarArchiveInputStream tis = new TarArchiveInputStream(tarStream)) {
+
+            TarArchiveEntry entry;
+            while ((entry = tis.getNextTarEntry()) != null) {
+              // Skip directories because we only want to copy metadata files from the container.
+              // Directory structure will be created automatically via 
+              // Files.createDirectories(outputPath.getParent()) when copying each file.
+              if (entry.isDirectory()) {
+                continue;
+              }
+
+              /* The tar entry names from the container include a prefix corresponding to the container's temporary warehouse directory, for example:
+               iceberg-test-1759245909247/iceberg_warehouse/ice_rest/ice_orc1/metadata/.00000-c37404c0-67a7-4355-8578-845952693a8c.metadata.json.crc
+              
+               This strips the first part (container temp dir) and takes the relative path inside the warehouse, 
+               so it maps correctly to the host warehouseDir: 
+               /tmp/iceberg-test-1759245909247/iceberg_warehouse/ice_rest/ice_orc1/metadata/.00000-c37404c0-67a7-4355-8578-845952693a8c.metadata.json.crc
+              
+               This ensures that files from the container are merged into the host warehouse without overwriting existing files. */
+              
+              String[] parts = entry.getName().split("/", 2);
+
+              // Defensive check: skip entries that don’t contain a relative path under the warehouse folder.
+              // Normally, all entries should have at least two parts: top-level container folder + relative path.
+              if (parts.length < 2) {
+                continue;
+              }
+
+              Path relativePath = Paths.get(parts[1]);
+              Path outputPath = warehouseDir.resolve(relativePath);
+
+              // Skip copying if the file already exists on the host.
+              // This ensures we merge new container files without overwriting existing ones.
+              if (Files.exists(outputPath)) {
+                continue;
+              }
+
+              Files.createDirectories(outputPath.getParent());
+              Files.copy(tis, outputPath);
+            }
+
+          } catch (Exception e) {
+            LOG.error("Warehouse folder sync failed: {}", e.getMessage(), e);
+          }
+
+          Thread.sleep(1000); // sync every 1 second
+        }
+      } catch (InterruptedException ignored) {
+        Thread.currentThread().interrupt();
+      }
+    });
+  }
+
   private void createWarehouseDir() {
     try {
       warehouseDir = Paths.get("/tmp", "iceberg-test-" + System.currentTimeMillis()).toAbsolutePath();