feat(test): add Spark integration tests for Hadoop table CRUD

Add cross-compatibility integration tests verifying CreateTable,
LoadTable, and CheckTableExists work between Go and Spark Hadoop
catalogs. Pre-create the hadoop-warehouse directory before Docker
compose to ensure runner ownership in CI.

Author: tanmayrauth

catalog/hadoop/hadoop_integration_test.go

@@ -26,6 +26,7 @@ import (
 	"strings"
 	"testing"
 
+	"github.com/apache/iceberg-go"
 	"github.com/apache/iceberg-go/catalog"
 	"github.com/apache/iceberg-go/catalog/hadoop"
 	"github.com/apache/iceberg-go/internal/recipe"
@@ -228,6 +229,82 @@ func (s *HadoopIntegrationSuite) TestDropTableThenListReflects() {
 	s.Equal("keep", tables[0][len(tables[0])-1])
 }
 
+// TestGoCreateSparkDescribe creates a table from Go, then asks Spark to
+// DESCRIBE it and verifies that Spark sees the expected columns.
+func (s *HadoopIntegrationSuite) TestGoCreateSparkDescribe() {
+	// Create namespace from Go so the directory is owned by the runner
+	// process and Go can create table subdirectories inside it.
+	err := s.cat.CreateNamespace(s.ctx, []string{"describe_ns"}, nil)
+	s.Require().NoError(err)
+
+	schema := iceberg.NewSchema(1,
+		iceberg.NestedField{ID: 1, Name: "id", Type: iceberg.PrimitiveTypes.Int64, Required: true},
+		iceberg.NestedField{ID: 2, Name: "name", Type: iceberg.PrimitiveTypes.String, Required: false},
+	)
+
+	tbl, err := s.cat.CreateTable(s.ctx, []string{"describe_ns", "go_created_table"}, schema)
+	s.Require().NoError(err)
+	s.NotNil(tbl)
+
+	output := s.sparkSQL("DESCRIBE TABLE hadoop_test.describe_ns.go_created_table")
+	s.Contains(output, "id")
+	s.Contains(output, "name")
+}
+
+// TestSparkCreateGoLoad creates a table from Spark, then loads it from
+// Go and verifies schema and metadata are consistent.
+func (s *HadoopIntegrationSuite) TestSparkCreateGoLoad() {
+	// Spark creates both namespace and table — Go only reads here.
+	s.sparkSQL("CREATE NAMESPACE IF NOT EXISTS hadoop_test.load_ns")
+	s.sparkSQL("CREATE TABLE hadoop_test.load_ns.spark_created_table (age INT, city STRING) USING iceberg")
+
+	tbl, err := s.cat.LoadTable(s.ctx, []string{"load_ns", "spark_created_table"})
+	s.Require().NoError(err)
+	s.NotNil(tbl)
+
+	fields := tbl.Schema().Fields()
+	fieldNames := make([]string, len(fields))
+	for i, f := range fields {
+		fieldNames[i] = f.Name
+	}
+
+	s.Contains(fieldNames, "age")
+	s.Contains(fieldNames, "city")
+}
+
+// TestGoCreateSparkSelect creates a table from Go, then runs a SELECT
+// from Spark to confirm the table is queryable (should return empty).
+func (s *HadoopIntegrationSuite) TestGoCreateSparkSelect() {
+	err := s.cat.CreateNamespace(s.ctx, []string{"select_ns"}, nil)
+	s.Require().NoError(err)
+
+	schema := iceberg.NewSchema(1,
+		iceberg.NestedField{ID: 1, Name: "value", Type: iceberg.PrimitiveTypes.Int32, Required: true},
+	)
+
+	_, err = s.cat.CreateTable(s.ctx, []string{"select_ns", "go_select_table"}, schema)
+	s.Require().NoError(err)
+
+	output := s.sparkSQL("SELECT * FROM hadoop_test.select_ns.go_select_table")
+	s.Contains(output, "value")
+}
+
+// TestGoCheckTableExistsForSparkTable verifies that CheckTableExists
+// returns true for a table created by Spark.
+func (s *HadoopIntegrationSuite) TestGoCheckTableExistsForSparkTable() {
+	// Spark creates both namespace and table — Go only checks existence.
+	s.sparkSQL("CREATE NAMESPACE IF NOT EXISTS hadoop_test.exists_ns")
+	s.sparkSQL("CREATE TABLE hadoop_test.exists_ns.spark_exists_table (x INT) USING iceberg")
+
+	exists, err := s.cat.CheckTableExists(s.ctx, []string{"exists_ns", "spark_exists_table"})
+	s.Require().NoError(err)
+	s.True(exists)
+
+	exists, err = s.cat.CheckTableExists(s.ctx, []string{"exists_ns", "no_such_table"})
+	s.Require().NoError(err)
+	s.False(exists)
+}
+
 func TestHadoopIntegration(t *testing.T) {
 	suite.Run(t, new(HadoopIntegrationSuite))
 }