zeroshade commented on code in PR #1244:
URL: https://github.com/apache/iceberg-go/pull/1244#discussion_r3521447790
##########
catalog/sql/sql.go:
##########
@@ -211,6 +246,10 @@ func NewCatalog(name string, db *sql.DB, dialect
SupportedDialect, props iceberg
return cat, cat.ensureTablesExist()
}
+ if err := cat.detectSchemaVersion(context.Background()); err != nil {
Review Comment:
[major] With `init_catalog_tables=false`, this path only detects the
existing schema and never calls `migrateV0Schema`, so `jdbc.schema-version=V1`
does not actually migrate an existing V0 catalog. Please run one
detect-and-maybe-migrate on open, and gate only `CreateSQLTables` behind
`init_catalog_tables`.
##########
catalog/sql/sql.go:
##########
@@ -247,12 +286,141 @@ func (c *Catalog) DropSQLTables(ctx context.Context)
error {
}
func (c *Catalog) ensureTablesExist() error {
- return c.CreateSQLTables(context.Background())
+ ctx := context.Background()
+ if err := c.CreateSQLTables(ctx); err != nil {
+ return err
+ }
+
+ return c.migrateV0Schema(ctx)
+}
+
+func (c *Catalog) detectSchemaVersion(ctx context.Context) error {
+ hasCol, err := c.icebergTypeColumnExists(ctx)
+ if err != nil {
+ return fmt.Errorf("detecting catalog schema version: %w", err)
+ }
+
+ if hasCol {
+ c.schemaVersion = schemaV1
+ } else {
+ c.schemaVersion = schemaV0
Review Comment:
[major] A missing `iceberg_tables` table is collapsed into `schemaV0`
because the column probe returns false for both legacy V0 and table-absent
catalogs. If a caller opens with `init_catalog_tables=false` and later calls
`CreateSQLTables`, the physical schema is V1 while this `Catalog` stays in V0
mode and writes omit `iceberg_type`. Please probe table existence separately
and either treat table-absent as fresh V1 or fail/document this path.
##########
catalog/sql/sql.go:
##########
@@ -1013,6 +1225,10 @@ func (c *Catalog) DropView(ctx context.Context,
identifier table.Identifier) err
// CheckViewExists returns true if a view exists in the catalog.
func (c *Catalog) CheckViewExists(ctx context.Context, identifier
table.Identifier) (bool, error) {
+ if c.isV0() {
+ return false, errViewsUnsupportedOnV0
Review Comment:
[minor] `CheckViewExists` should return `(false, nil)` on V0 rather than an
unsupported-view error, mirroring `CheckTableExists` existence-check semantics.
Consider making `ListViews` return empty on V0 as well.
##########
catalog/sql/sql.go:
##########
@@ -247,12 +286,141 @@ func (c *Catalog) DropSQLTables(ctx context.Context)
error {
}
func (c *Catalog) ensureTablesExist() error {
- return c.CreateSQLTables(context.Background())
+ ctx := context.Background()
+ if err := c.CreateSQLTables(ctx); err != nil {
+ return err
+ }
+
+ return c.migrateV0Schema(ctx)
+}
+
+func (c *Catalog) detectSchemaVersion(ctx context.Context) error {
+ hasCol, err := c.icebergTypeColumnExists(ctx)
+ if err != nil {
+ return fmt.Errorf("detecting catalog schema version: %w", err)
+ }
+
+ if hasCol {
+ c.schemaVersion = schemaV1
+ } else {
+ c.schemaVersion = schemaV0
+ }
+
+ return nil
+}
+
+func (c *Catalog) migrateV0Schema(ctx context.Context) error {
+ hasCol, err := c.icebergTypeColumnExists(ctx)
+ if err != nil {
+ return fmt.Errorf("V0 schema migration: column probe failed:
%w", err)
+ }
+ if hasCol {
+ c.schemaVersion = schemaV1
+
+ return nil
+ }
+
+ if !strings.EqualFold(c.props.Get(SchemaVersionKey, ""),
SchemaVersionV1) {
Review Comment:
[major] `strings.EqualFold(..., "V1")` lets lowercase `v1` trigger the
irreversible ALTER, which diverges from Java's case-sensitive handling. Please
compare exactly to `SchemaVersionV1`, or explicitly document and test the
intended case-insensitive behavior.
##########
catalog/sql/sql.go:
##########
@@ -247,12 +286,141 @@ func (c *Catalog) DropSQLTables(ctx context.Context)
error {
}
func (c *Catalog) ensureTablesExist() error {
- return c.CreateSQLTables(context.Background())
+ ctx := context.Background()
+ if err := c.CreateSQLTables(ctx); err != nil {
+ return err
+ }
+
+ return c.migrateV0Schema(ctx)
+}
+
+func (c *Catalog) detectSchemaVersion(ctx context.Context) error {
+ hasCol, err := c.icebergTypeColumnExists(ctx)
+ if err != nil {
+ return fmt.Errorf("detecting catalog schema version: %w", err)
+ }
+
+ if hasCol {
+ c.schemaVersion = schemaV1
+ } else {
+ c.schemaVersion = schemaV0
+ }
+
+ return nil
+}
+
+func (c *Catalog) migrateV0Schema(ctx context.Context) error {
+ hasCol, err := c.icebergTypeColumnExists(ctx)
+ if err != nil {
+ return fmt.Errorf("V0 schema migration: column probe failed:
%w", err)
+ }
+ if hasCol {
+ c.schemaVersion = schemaV1
+
+ return nil
+ }
+
+ if !strings.EqualFold(c.props.Get(SchemaVersionKey, ""),
SchemaVersionV1) {
+ log.Printf("WARNING: iceberg_tables is on the legacy V0 schema
(no iceberg_type column); "+
+ "set %q=%q to migrate to V1. Operating in V0-compatible
mode (column-free queries).",
+ SchemaVersionKey, SchemaVersionV1)
+ c.schemaVersion = schemaV0
+
+ return nil
+ }
+
+ ddl, err := addIcebergTypeColumnDDL(c.db.Dialect().Name())
+ if err != nil {
+ return err
+ }
+
+ if _, err := c.db.ExecContext(ctx, ddl); err != nil {
+ has, reprobeErr := c.icebergTypeColumnExists(ctx)
+ if has {
+ c.schemaVersion = schemaV1
+
+ return nil
+ }
+ if reprobeErr != nil {
+ return fmt.Errorf("migration of V0 schema failed: %w
(re-probe also failed: %v)", err, reprobeErr)
+ }
+
+ return fmt.Errorf("migration of V0 schema failed: %w", err)
+ }
+
+ c.schemaVersion = schemaV1
+
+ return nil
+}
+
+func addIcebergTypeColumnDDL(name dialect.Name) (string, error) {
+ switch name {
+ case dialect.PG, dialect.SQLite:
+ return "ALTER TABLE iceberg_tables ADD COLUMN iceberg_type
VARCHAR", nil
+ case dialect.MySQL:
+ return "ALTER TABLE iceberg_tables ADD COLUMN iceberg_type
VARCHAR(255)", nil
+ case dialect.MSSQL:
+ return "ALTER TABLE iceberg_tables ADD iceberg_type
VARCHAR(255)", nil
+ case dialect.Oracle:
+ return "ALTER TABLE iceberg_tables ADD (iceberg_type
VARCHAR2(255))", nil
+ default:
+ return "", fmt.Errorf("unsupported dialect for V0 migration:
%s", name)
+ }
+}
+
+func icebergTypeColumnExistsQuery(name dialect.Name) (string, error) {
+ switch name {
+ case dialect.PG:
+ return `SELECT 1 FROM information_schema.columns
+ WHERE table_schema = current_schema()
+ AND table_name = 'iceberg_tables'
+ AND column_name = 'iceberg_type' LIMIT 1`, nil
+ case dialect.MySQL:
+ return `SELECT 1 FROM information_schema.columns
+ WHERE table_schema = DATABASE()
+ AND table_name = 'iceberg_tables'
+ AND column_name = 'iceberg_type' LIMIT 1`, nil
+ case dialect.MSSQL:
+ return `SELECT TOP 1 1 FROM information_schema.columns
+ WHERE table_schema = SCHEMA_NAME()
+ AND table_name = 'iceberg_tables'
+ AND column_name = 'iceberg_type'`, nil
+ case dialect.SQLite:
+ return `SELECT 1 FROM pragma_table_info('iceberg_tables')
+ WHERE name = 'iceberg_type' LIMIT 1`, nil
+ case dialect.Oracle:
+ return `SELECT 1 FROM user_tab_columns
+ WHERE UPPER(table_name) = 'ICEBERG_TABLES'
+ AND UPPER(column_name) = 'ICEBERG_TYPE'`, nil
+ default:
+ return "", fmt.Errorf("unsupported dialect for V0 migration:
%s", name)
+ }
+}
+
+func (c *Catalog) icebergTypeColumnExists(ctx context.Context) (bool, error) {
+ query, err := icebergTypeColumnExistsQuery(c.db.Dialect().Name())
+ if err != nil {
+ return false, err
+ }
+
+ var exists int
+ if err := c.db.QueryRowContext(ctx, query).Scan(&exists); err != nil {
Review Comment:
[minor] The schema probe bypasses `withReadTx`, unlike the other read paths
in this catalog. Please wrap this in `withReadTx` as well, or document why the
schema-version probe intentionally runs outside the transaction helper.
##########
catalog/sql/sql.go:
##########
@@ -61,8 +62,37 @@ const (
DialectKey = "sql.dialect"
DriverKey = "sql.driver"
initCatalogTablesKey = "init_catalog_tables"
+
+ // SchemaVersionKey opts in to migrating a legacy V0 catalog (one that
lacks
+ // the iceberg_type column) up to the V1 schema.
+ //
+ // When set to SchemaVersionV1 the catalog issues the one-way ALTER
TABLE
+ // that adds the iceberg_type column; otherwise it stays on V0 and emits
+ // column-free queries that never reference iceberg_type.
+ SchemaVersionKey = "jdbc.schema-version"
+)
+
+const (
+ SchemaVersionV1 = "V1"
)
+// schemaVer tracks which physical layout of the iceberg_tables table the
catalog is operating against.
+//
+// schemaV1 has the iceberg_type column. schemaV0 is the legacy JDBC default
that lacks the column entirely.
+// Queries against a V0 catalog must never reference iceberg_type
+//
+// The zero value is schemaV1
+type schemaVer int
+
+const (
+ schemaV1 schemaVer = iota
Review Comment:
[minor] The zero value of `schemaVer` is currently V1, which is the less
safe default if construction or probing is skipped. Please make `schemaV0` the
iota zero value and set V1 only after a successful probe or migration.
##########
catalog/sql/sql.go:
##########
@@ -247,12 +286,141 @@ func (c *Catalog) DropSQLTables(ctx context.Context)
error {
}
func (c *Catalog) ensureTablesExist() error {
- return c.CreateSQLTables(context.Background())
+ ctx := context.Background()
+ if err := c.CreateSQLTables(ctx); err != nil {
+ return err
+ }
+
+ return c.migrateV0Schema(ctx)
+}
+
+func (c *Catalog) detectSchemaVersion(ctx context.Context) error {
+ hasCol, err := c.icebergTypeColumnExists(ctx)
+ if err != nil {
+ return fmt.Errorf("detecting catalog schema version: %w", err)
+ }
+
+ if hasCol {
+ c.schemaVersion = schemaV1
+ } else {
+ c.schemaVersion = schemaV0
+ }
+
+ return nil
+}
+
+func (c *Catalog) migrateV0Schema(ctx context.Context) error {
+ hasCol, err := c.icebergTypeColumnExists(ctx)
+ if err != nil {
+ return fmt.Errorf("V0 schema migration: column probe failed:
%w", err)
+ }
+ if hasCol {
+ c.schemaVersion = schemaV1
+
+ return nil
+ }
+
+ if !strings.EqualFold(c.props.Get(SchemaVersionKey, ""),
SchemaVersionV1) {
+ log.Printf("WARNING: iceberg_tables is on the legacy V0 schema
(no iceberg_type column); "+
+ "set %q=%q to migrate to V1. Operating in V0-compatible
mode (column-free queries).",
+ SchemaVersionKey, SchemaVersionV1)
+ c.schemaVersion = schemaV0
+
+ return nil
+ }
+
+ ddl, err := addIcebergTypeColumnDDL(c.db.Dialect().Name())
+ if err != nil {
+ return err
+ }
+
+ if _, err := c.db.ExecContext(ctx, ddl); err != nil {
+ has, reprobeErr := c.icebergTypeColumnExists(ctx)
+ if has {
+ c.schemaVersion = schemaV1
+
+ return nil
+ }
+ if reprobeErr != nil {
+ return fmt.Errorf("migration of V0 schema failed: %w
(re-probe also failed: %v)", err, reprobeErr)
+ }
+
+ return fmt.Errorf("migration of V0 schema failed: %w", err)
+ }
+
+ c.schemaVersion = schemaV1
+
+ return nil
+}
+
+func addIcebergTypeColumnDDL(name dialect.Name) (string, error) {
+ switch name {
+ case dialect.PG, dialect.SQLite:
+ return "ALTER TABLE iceberg_tables ADD COLUMN iceberg_type
VARCHAR", nil
+ case dialect.MySQL:
+ return "ALTER TABLE iceberg_tables ADD COLUMN iceberg_type
VARCHAR(255)", nil
+ case dialect.MSSQL:
+ return "ALTER TABLE iceberg_tables ADD iceberg_type
VARCHAR(255)", nil
+ case dialect.Oracle:
+ return "ALTER TABLE iceberg_tables ADD (iceberg_type
VARCHAR2(255))", nil
+ default:
+ return "", fmt.Errorf("unsupported dialect for V0 migration:
%s", name)
+ }
+}
+
+func icebergTypeColumnExistsQuery(name dialect.Name) (string, error) {
+ switch name {
+ case dialect.PG:
+ return `SELECT 1 FROM information_schema.columns
+ WHERE table_schema = current_schema()
+ AND table_name = 'iceberg_tables'
+ AND column_name = 'iceberg_type' LIMIT 1`, nil
+ case dialect.MySQL:
+ return `SELECT 1 FROM information_schema.columns
+ WHERE table_schema = DATABASE()
+ AND table_name = 'iceberg_tables'
+ AND column_name = 'iceberg_type' LIMIT 1`, nil
+ case dialect.MSSQL:
+ return `SELECT TOP 1 1 FROM information_schema.columns
+ WHERE table_schema = SCHEMA_NAME()
+ AND table_name = 'iceberg_tables'
+ AND column_name = 'iceberg_type'`, nil
+ case dialect.SQLite:
+ return `SELECT 1 FROM pragma_table_info('iceberg_tables')
+ WHERE name = 'iceberg_type' LIMIT 1`, nil
+ case dialect.Oracle:
+ return `SELECT 1 FROM user_tab_columns
Review Comment:
[nit] The Oracle column probe should also limit to one row, e.g. add `AND
ROWNUM = 1`, to match the other dialect probes' bounded existence checks.
##########
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 {
+ cat, err := catalog.Load(context.Background(), "default",
iceberg.Properties{
+ "uri": s.catalogUri(),
+ sqlcat.DriverKey: sqliteshim.ShimName,
+ sqlcat.DialectKey: string(sqlcat.SQLite),
+ "type": "sql",
+ "warehouse": "file://" + s.warehouse,
+ })
+ s.Require().NoError(err)
+
+ return cat.(*sqlcat.Catalog)
+}
+
+func (s *SqliteCatalogTestSuite) icebergTypeColumnAbsent(sqldb *sql.DB) {
+ 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 remain absent
on a V0 catalog")
+}
+
+func (s *SqliteCatalogTestSuite) TestV0SchemaFullLifecycleWithoutOptIn() {
+ sqldb := s.getDB()
+ s.confirmNoTables(sqldb)
+ s.createLegacyV0Catalog(sqldb)
+
+ cat := s.loadCatalogV0NoOptIn()
+ ctx := context.Background()
+
+ s.icebergTypeColumnAbsent(sqldb)
+
+ tblID := s.randomTableIdentifier()
+ ns := catalog.NamespaceFromIdent(tblID)
+ s.Require().NoError(cat.CreateNamespace(ctx, ns, nil))
+
+ created, err := cat.CreateTable(ctx, tblID, tableSchemaNested)
+ s.Require().NoError(err, "CreateTable must work on a column-less V0
schema")
+ s.Equal(tblID, created.Identifier())
+
+ loaded, err := cat.LoadTable(ctx, tblID)
+ s.Require().NoError(err, "LoadTable must work on a column-less V0
schema")
+ s.Equal(tblID, loaded.Identifier())
+
+ var listed []table.Identifier
+ for ident, iterErr := range cat.ListTables(ctx, ns) {
+ s.Require().NoError(iterErr)
+ listed = append(listed, ident)
+ }
+ s.Require().Len(listed, 1, "ListTables must work on a column-less V0
schema")
+ s.Equal(tblID, listed[0])
+
+ tx := loaded.NewTransaction()
+
s.Require().NoError(tx.SetProperties(iceberg.Properties{"v0.commit.test":
"ok"}))
+ _, err = tx.Commit(ctx)
+ s.Require().NoError(err, "CommitTable must work on a column-less V0
schema")
+
+ 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 work on a column-less V0
schema")
+ s.Equal(toID, renamed.Identifier())
+
+ s.Require().NoError(cat.DropTable(ctx, toID), "DropTable must work on a
column-less V0 schema")
+ _, err = cat.LoadTable(ctx, toID)
+ s.ErrorIs(err, catalog.ErrNoSuchTable)
+
+ s.icebergTypeColumnAbsent(sqldb)
+}
+
+func (s *SqliteCatalogTestSuite) TestV0SchemaViewOperationsUnsupported() {
+ sqldb := s.getDB()
+ s.confirmNoTables(sqldb)
+ s.createLegacyV0Catalog(sqldb)
+
+ cat := s.loadCatalogV0NoOptIn()
+ ctx := context.Background()
+
+ nsName := databaseName()
+ s.Require().NoError(cat.CreateNamespace(ctx, []string{nsName}, nil))
+
+ schema := iceberg.NewSchema(1, iceberg.NestedField{
+ ID: 1, Name: "id", Type: iceberg.PrimitiveTypes.Int32,
Required: true,
+ })
+ viewID := []string{nsName, tableName()}
+
+ err := cat.CreateView(ctx, viewID, schema, "SELECT 1", nil)
+ s.ErrorContains(err, "V0", "CreateView must be rejected on a V0
catalog")
+
+ _, err = cat.CheckViewExists(ctx, viewID)
+ s.ErrorContains(err, "V0", "CheckViewExists must be rejected on a V0
catalog")
+
+ _, err = cat.LoadView(ctx, viewID)
+ s.ErrorContains(err, "V0", "LoadView must be rejected on a V0 catalog")
+
+ err = cat.DropView(ctx, viewID)
+ s.ErrorContains(err, "V0", "DropView must be rejected on a V0 catalog")
+
+ for _, listErr := range cat.ListViews(ctx, []string{nsName}) {
Review Comment:
[nit] This `ListViews` assertion can pass vacuously if the iterator yields
no values. Please assert that the iterator produced the expected error (or
collect and assert one error) so the V0 behavior is actually exercised.
--
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]
