badalprasadsingh commented on code in PR #1244:
URL: https://github.com/apache/iceberg-go/pull/1244#discussion_r3530952751
##########
catalog/sql/sql_test.go:
##########
@@ -322,6 +336,358 @@ func (s *SqliteCatalogTestSuite)
TestCreationAllTablesExist() {
s.confirmTablesExist(sqldb)
}
+func (s *SqliteCatalogTestSuite) TestV0SchemaFromIcebergIsMigrated() {
+ sqldb := s.getDB()
+ s.confirmNoTables(sqldb)
+
+ _, err := sqldb.Exec(`CREATE TABLE "iceberg_tables" (
+ "catalog_name" VARCHAR NOT NULL,
+ "table_namespace" VARCHAR NOT NULL,
+ "table_name" VARCHAR NOT NULL,
+ "metadata_location" VARCHAR,
+ "previous_metadata_location" VARCHAR,
+ PRIMARY KEY ("catalog_name", "table_namespace", "table_name"))`)
+ s.Require().NoError(err)
+ _, err = sqldb.Exec(`CREATE TABLE "iceberg_namespace_properties" (
+ "catalog_name" VARCHAR NOT NULL,
+ "namespace" VARCHAR NOT NULL,
+ "property_key" VARCHAR NOT NULL,
+ "property_value" VARCHAR,
+ PRIMARY KEY ("catalog_name", "namespace", "property_key"))`)
+ s.Require().NoError(err)
+
+ const (
+ catName = "default"
+ nsName = "test_v0_ns"
+ tblName = "test_v0_tbl"
+ metaLoc = "file:///does/not/matter/00000-v0.metadata.json"
+ )
+ _, err = sqldb.Exec(
+ `INSERT INTO "iceberg_namespace_properties" `+
+ `("catalog_name", "namespace", "property_key",
"property_value") VALUES (?, ?, ?, ?)`,
+ catName, nsName, "exists", "true")
+ s.Require().NoError(err)
+ _, err = sqldb.Exec(
+ `INSERT INTO "iceberg_tables" `+
+ `("catalog_name", "table_namespace", "table_name",
"metadata_location") VALUES (?, ?, ?, ?)`,
+ catName, nsName, tblName, metaLoc)
+ s.Require().NoError(err)
+
+ _ = s.loadCatalogWithV1Migration()
+
+ columnPresent := func() bool {
+ var present int
+ qErr := sqldb.QueryRow(
+ `SELECT 1 FROM pragma_table_info('iceberg_tables')
WHERE name = 'iceberg_type'`,
+ ).Scan(&present)
+ s.Require().NoError(qErr, "migration must add iceberg_type
column")
+
+ return present == 1
+ }
+ s.True(columnPresent(), "migration must add iceberg_type column")
+
+ rowType := func() sql.NullString {
+ var got sql.NullString
+ qErr := sqldb.QueryRow(
+ `SELECT iceberg_type FROM iceberg_tables WHERE
catalog_name = ? AND table_namespace = ? AND table_name = ?`,
+ catName, nsName, tblName,
+ ).Scan(&got)
+ s.Require().NoError(qErr)
+
+ return got
+ }
+ s.False(rowType().Valid, "pre-V0 row should keep iceberg_type IS NULL
after migration")
+
+ // Re-opening with the same V1 opt-in must be idempotent: the column
stays
+ // present and the legacy row's iceberg_type is left NULL (not re-added
or stamped).
+ _ = s.loadCatalogWithV1Migration()
+ s.True(columnPresent(), "column must remain present after a second
open")
+ s.False(rowType().Valid, "legacy row must remain NULL after a second
open")
+}
+
+func (s *SqliteCatalogTestSuite) TestV0SchemaNotMigratedWithoutOptIn() {
+ sqldb := s.getDB()
+ s.confirmNoTables(sqldb)
+
+ _, err := sqldb.Exec(`CREATE TABLE "iceberg_tables" (
+ "catalog_name" VARCHAR NOT NULL,
+ "table_namespace" VARCHAR NOT NULL,
+ "table_name" VARCHAR NOT NULL,
+ "metadata_location" VARCHAR,
+ "previous_metadata_location" VARCHAR,
+ PRIMARY KEY ("catalog_name", "table_namespace", "table_name"))`)
+ s.Require().NoError(err)
+ _, err = sqldb.Exec(`CREATE TABLE "iceberg_namespace_properties" (
+ "catalog_name" VARCHAR NOT NULL,
+ "namespace" VARCHAR NOT NULL,
+ "property_key" VARCHAR NOT NULL,
+ "property_value" VARCHAR,
+ PRIMARY KEY ("catalog_name", "namespace", "property_key"))`)
+ s.Require().NoError(err)
+
+ // Without jdbc.schema-version=V1, opening a legacy V0 catalog must NOT
issue
+ // the one-way ALTER; the column stays absent.
+ _ = s.loadCatalogForTableCreation()
+
+ var present int
+ err = sqldb.QueryRow(
+ `SELECT 1 FROM pragma_table_info('iceberg_tables') WHERE name =
'iceberg_type'`,
+ ).Scan(&present)
+ s.ErrorIs(err, sql.ErrNoRows, "iceberg_type column must not be added
without the V1 opt-in")
+}
+
+func (s *SqliteCatalogTestSuite)
TestV0SchemaListAndDropAcceptNullIcebergType() {
+ sqldb := s.getDB()
+ s.confirmNoTables(sqldb)
+
+ _, err := sqldb.Exec(`CREATE TABLE "iceberg_tables" (
+ "catalog_name" VARCHAR NOT NULL,
+ "table_namespace" VARCHAR NOT NULL,
+ "table_name" VARCHAR NOT NULL,
+ "metadata_location" VARCHAR,
+ "previous_metadata_location" VARCHAR,
+ PRIMARY KEY ("catalog_name", "table_namespace", "table_name"))`)
+ s.Require().NoError(err)
+ _, err = sqldb.Exec(`CREATE TABLE "iceberg_namespace_properties" (
+ "catalog_name" VARCHAR NOT NULL,
+ "namespace" VARCHAR NOT NULL,
+ "property_key" VARCHAR NOT NULL,
+ "property_value" VARCHAR,
+ PRIMARY KEY ("catalog_name", "namespace", "property_key"))`)
+ s.Require().NoError(err)
+
+ const (
+ catName = "default"
+ nsName = "legacy_ns"
+ tblName = "legacy_tbl"
+ metaLoc = "file:///legacy/00000-v0.metadata.json"
+ )
+ _, err = sqldb.Exec(
+ `INSERT INTO "iceberg_namespace_properties" `+
+ `("catalog_name", "namespace", "property_key",
"property_value") VALUES (?, ?, ?, ?)`,
+ catName, nsName, "exists", "true")
+ s.Require().NoError(err)
+ _, err = sqldb.Exec(
+ `INSERT INTO "iceberg_tables" `+
+ `("catalog_name", "table_namespace", "table_name",
"metadata_location") VALUES (?, ?, ?, ?)`,
+ catName, nsName, tblName, metaLoc)
+ s.Require().NoError(err)
+
+ // Migrate to V1: this adds the iceberg_type column and leaves the
legacy row
+ // with iceberg_type IS NULL, which the read path must still tolerate.
+ cat := s.loadCatalogWithV1Migration()
+ ctx := context.Background()
+
+ var listed []table.Identifier
+ for ident, err := range cat.ListTables(ctx, table.Identifier{nsName}) {
+ s.Require().NoError(err)
+ listed = append(listed, ident)
+ }
+ s.Require().Len(listed, 1, "ListTables must return rows with
iceberg_type IS NULL")
+ s.Equal(tblName, listed[0][len(listed[0])-1])
+
+ s.Require().NoError(cat.DropTable(ctx, table.Identifier{nsName,
tblName}))
+
+ var remaining int
+ err = sqldb.QueryRow(
+ `SELECT COUNT(*) FROM iceberg_tables WHERE catalog_name = ? AND
table_namespace = ? AND table_name = ?`,
+ catName, nsName, tblName,
+ ).Scan(&remaining)
+ s.Require().NoError(err)
+ s.Equal(0, remaining, "DropTable must delete rows with iceberg_type IS
NULL")
+}
+
+func (s *SqliteCatalogTestSuite)
TestV0SchemaLoadCommitRenameAcceptNullIcebergType() {
+ sqldb := s.getDB()
+
+ cat := s.getCatalogSqlite()
+ ctx := context.Background()
+
+ // This is a proxy for a genuine V0 row: we create the row at V1 (so the
+ // iceberg_type column already exists) and then manually NULL it,
rather than
+ // starting from a truly column-less V0 schema migrated up. It
exercises the
+ // same NULL-tolerant predicate the read/commit/rename paths rely on.
+
+ tblID := s.randomTableIdentifier()
+ ns := catalog.NamespaceFromIdent(tblID)
+ s.Require().NoError(cat.CreateNamespace(ctx, ns, nil))
+
+ tbl, err := cat.CreateTable(ctx, tblID, tableSchemaNested)
+ s.Require().NoError(err)
+ s.Equal(tblID, tbl.Identifier())
+
+ nullIcebergType := func(ident table.Identifier) {
+ res, exErr := sqldb.Exec(
+ `UPDATE iceberg_tables SET iceberg_type = NULL `+
+ `WHERE catalog_name = ? AND table_namespace = ?
AND table_name = ?`,
+ cat.Name(),
+ strings.Join(catalog.NamespaceFromIdent(ident), "."),
+ catalog.TableNameFromIdent(ident),
+ )
+ s.Require().NoError(exErr)
+ n, raErr := res.RowsAffected()
+ s.Require().NoError(raErr)
+ s.Require().Equal(int64(1), n, "test setup: expected exactly
one row to be NULLed")
+ }
+ nullIcebergType(tblID)
+
+ loaded, err := cat.LoadTable(ctx, tblID)
+ s.Require().NoError(err, "LoadTable must accept rows with iceberg_type
IS NULL")
+ s.Equal(tblID, loaded.Identifier())
+
+ tx := loaded.NewTransaction()
+
s.Require().NoError(tx.SetProperties(iceberg.Properties{"v0.commit.test":
"ok"}))
+ _, err = tx.Commit(ctx)
+ s.Require().NoError(err, "CommitTable must accept rows with
iceberg_type IS NULL")
+
+ var icebergType sql.NullString
+ err = sqldb.QueryRow(
+ `SELECT iceberg_type FROM iceberg_tables WHERE catalog_name = ?
AND table_namespace = ? AND table_name = ?`,
+ cat.Name(),
+ strings.Join(ns, "."),
+ catalog.TableNameFromIdent(tblID),
+ ).Scan(&icebergType)
+ s.Require().NoError(err)
+ s.True(icebergType.Valid, "CommitTable on a V0 row must stamp
iceberg_type back to TABLE")
+ s.Equal(sqlcat.TableType, icebergType.String)
+
+ nullIcebergType(tblID)
+
+ toID := s.randomTableIdentifier()
+ toNs := catalog.NamespaceFromIdent(toID)
+ s.Require().NoError(cat.CreateNamespace(ctx, toNs, nil))
+
+ renamed, err := cat.RenameTable(ctx, tblID, toID)
+ s.Require().NoError(err, "RenameTable must accept rows with
iceberg_type IS NULL")
+ s.Equal(toID, renamed.Identifier())
+}
+
+func (s *SqliteCatalogTestSuite) createLegacyV0Catalog(sqldb *sql.DB) {
+ _, err := sqldb.Exec(`CREATE TABLE "iceberg_tables" (
+ "catalog_name" VARCHAR NOT NULL,
+ "table_namespace" VARCHAR NOT NULL,
+ "table_name" VARCHAR NOT NULL,
+ "metadata_location" VARCHAR,
+ "previous_metadata_location" VARCHAR,
+ PRIMARY KEY ("catalog_name", "table_namespace", "table_name"))`)
+ s.Require().NoError(err)
+ _, err = sqldb.Exec(`CREATE TABLE "iceberg_namespace_properties" (
+ "catalog_name" VARCHAR NOT NULL,
+ "namespace" VARCHAR NOT NULL,
+ "property_key" VARCHAR NOT NULL,
+ "property_value" VARCHAR,
+ PRIMARY KEY ("catalog_name", "namespace", "property_key"))`)
+ s.Require().NoError(err)
+}
+
+func (s *SqliteCatalogTestSuite) loadCatalogV0NoOptIn() *sqlcat.Catalog {
Review Comment:
Done.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
