laskoviymishka commented on code in PR #1244:
URL: https://github.com/apache/iceberg-go/pull/1244#discussion_r3455856277
##########
catalog/sql/sql.go:
##########
@@ -247,7 +247,76 @@ 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) migrateV0Schema(ctx context.Context) error {
+ hasCol, probeErr := c.icebergTypeColumnExists(ctx)
+ if probeErr != nil {
+ log.Printf("WARNING: skipping V0 schema migration; iceberg_type
column probe failed: %v", probeErr)
+
+ return nil
+ }
+ if hasCol {
+ return nil
+ }
+
+ if _, err := c.db.ExecContext(ctx, "ALTER TABLE iceberg_tables ADD
COLUMN iceberg_type VARCHAR(5)"); err != nil {
+ has, reprobeErr := c.icebergTypeColumnExists(ctx)
+ if has {
+ return nil
+ }
+ if reprobeErr != nil {
+ log.Printf("WARNING: V0 schema migration re-probe
failed after ALTER error: %v", reprobeErr)
+ }
+
+ return fmt.Errorf("migration of V0 schema failed: %w", err)
+ }
+
+ return nil
+}
+
+func (c *Catalog) icebergTypeColumnExists(ctx context.Context) (bool, error) {
+ var query string
+ switch dialectName := c.db.Dialect().Name().String(); dialectName {
+ case "pg":
Review Comment:
The Postgres probe doesn't scope to the current schema, and the `mssql` case
below has the same gap. `information_schema.columns` spans every schema the
user can see, so if another schema has an `iceberg_tables` that already carries
`iceberg_type`, this returns true, we skip the migration, and the first real
query against our table fails. Multi-schema Postgres is common, and the JDBC
catalog supports a configurable schema. I'd add `AND table_schema =
current_schema()` here and `AND table_schema = SCHEMA_NAME()` to the MSSQL case
— MySQL already gets this right with `table_schema = DATABASE()`.
##########
catalog/sql/sql.go:
##########
@@ -247,7 +247,76 @@ 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) migrateV0Schema(ctx context.Context) error {
Review Comment:
This is the piece I'd most want to settle before merge. The migration runs
on every open where `init_catalog_tables` is set (the default), so we issue
one-way DDL against what may be a catalog shared with Java or PyIceberg, with
no opt-out.
Java made this an explicit choice — `JdbcCatalog` only migrates when
`jdbc.schema-version=V1` is set, otherwise it warns and stays on V0. I'd mirror
that: take a `jdbc.schema-version` property (same key as Java, for interop),
run the ALTER only when it's `V1`, and otherwise lean on the NULL-tolerant read
path this PR already adds. There's also no exported constant for that knob the
way `init_catalog_tables` has one. wdyt?
##########
catalog/sql/sql.go:
##########
@@ -247,7 +247,76 @@ 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) migrateV0Schema(ctx context.Context) error {
+ hasCol, probeErr := c.icebergTypeColumnExists(ctx)
+ if probeErr != nil {
Review Comment:
Swallowing the probe error and returning nil means `NewCatalog` hands back a
catalog that looks fine, then the first table op fails with a cryptic `no such
column: iceberg_type` far from the real cause. The `default` case in the probe
gets silently no-op'd the same way. I'd return the wrapped error here
(`fmt.Errorf("V0 schema migration: column probe failed: %w", probeErr)`) so we
fail fast at construction; if the goal is to let read-only users who can't
probe still connect, I'd make that an explicit option rather than the silent
default.
(Minor, while here: `log.Printf` writes to the default logger and bypasses
whatever logging the caller configured — consistent with `PurgeTable` today,
but worth keeping in mind.)
##########
catalog/sql/sql.go:
##########
@@ -247,7 +247,76 @@ 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) migrateV0Schema(ctx context.Context) error {
+ hasCol, probeErr := c.icebergTypeColumnExists(ctx)
+ if probeErr != nil {
+ log.Printf("WARNING: skipping V0 schema migration; iceberg_type
column probe failed: %v", probeErr)
+
+ return nil
+ }
+ if hasCol {
+ return nil
+ }
+
+ if _, err := c.db.ExecContext(ctx, "ALTER TABLE iceberg_tables ADD
COLUMN iceberg_type VARCHAR(5)"); err != nil {
+ has, reprobeErr := c.icebergTypeColumnExists(ctx)
+ if has {
+ return nil
+ }
+ if reprobeErr != nil {
+ log.Printf("WARNING: V0 schema migration re-probe
failed after ALTER error: %v", reprobeErr)
+ }
+
+ return fmt.Errorf("migration of V0 schema failed: %w", err)
+ }
+
+ return nil
+}
+
+func (c *Catalog) icebergTypeColumnExists(ctx context.Context) (bool, error) {
+ var query string
+ switch dialectName := c.db.Dialect().Name().String(); dialectName {
+ case "pg":
+ query = `SELECT 1 FROM information_schema.columns
+ WHERE table_name = 'iceberg_tables'
+ AND column_name = 'iceberg_type' LIMIT 1`
+ case "mysql":
+ query = `SELECT 1 FROM information_schema.columns
+ WHERE table_schema = DATABASE()
+ AND table_name = 'iceberg_tables'
+ AND column_name = 'iceberg_type' LIMIT 1`
+ case "mssql":
+ query = `SELECT TOP 1 1 FROM information_schema.columns
+ WHERE table_name = 'iceberg_tables'
+ AND column_name = 'iceberg_type'`
+ case "sqlite":
+ query = `SELECT 1 FROM pragma_table_info('iceberg_tables')
+ WHERE name = 'iceberg_type' LIMIT 1`
+ case "oracle":
+ query = `SELECT 1 FROM user_tab_columns
+ WHERE UPPER(table_name) = 'ICEBERG_TABLES'
+ AND UPPER(column_name) = 'ICEBERG_TYPE'`
+ default:
+ return false, fmt.Errorf("unsupported dialect for V0 migration:
%s", dialectName)
+ }
+ var dummy int
Review Comment:
`dummy` undersells it — the scanned value is a real existence sentinel. `var
exists int` reads better, or skip the scan target entirely by probing with
`COUNT(*)` into an int and checking `> 0`.
##########
catalog/sql/sql.go:
##########
@@ -247,7 +247,76 @@ 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) migrateV0Schema(ctx context.Context) error {
+ hasCol, probeErr := c.icebergTypeColumnExists(ctx)
+ if probeErr != nil {
+ log.Printf("WARNING: skipping V0 schema migration; iceberg_type
column probe failed: %v", probeErr)
+
+ return nil
+ }
+ if hasCol {
+ return nil
+ }
+
+ if _, err := c.db.ExecContext(ctx, "ALTER TABLE iceberg_tables ADD
COLUMN iceberg_type VARCHAR(5)"); err != nil {
+ has, reprobeErr := c.icebergTypeColumnExists(ctx)
+ if has {
+ return nil
+ }
+ if reprobeErr != nil {
+ log.Printf("WARNING: V0 schema migration re-probe
failed after ALTER error: %v", reprobeErr)
+ }
+
+ return fmt.Errorf("migration of V0 schema failed: %w", err)
+ }
+
+ return nil
+}
+
+func (c *Catalog) icebergTypeColumnExists(ctx context.Context) (bool, error) {
+ var query string
+ switch dialectName := c.db.Dialect().Name().String(); dialectName {
Review Comment:
Small robustness thing: switching on `c.db.Dialect().Name().String()`
compares against bun's internal strings and leaks one back out in the `default`
error (e.g. `pg`, not the public `postgres`). bun exports the typed
`dialect.Name` constants (`dialect.PG`, `dialect.SQLite`, …) — switching on
those instead is sturdier against name drift and lets the `exhaustive` linter
actually check this switch.
##########
catalog/sql/sql.go:
##########
@@ -349,11 +418,11 @@ func (c *Catalog) CommitTable(ctx context.Context, ident
table.Identifier, reqs
CatalogName: c.name,
TableNamespace: strings.Join(ns, "."),
TableName: tblName,
- IcebergType: TableType,
+ IcebergType:
sql.NullString{String: TableType, Valid: true},
Review Comment:
Heads up — because this UPDATE is driven off the model struct, bun puts
every non-PK field in the SET clause, so we write `iceberg_type = 'TABLE'` back
onto the row on every commit. Java's V1 commit only sets
`metadata_location`/`previous_metadata_location` and leaves `iceberg_type`
alone (it's WHERE-only there). It's harmless — NULL and `'TABLE'` read
identically through the new predicate — but it's an intentional-looking
divergence the test asserts as correct, so either a one-line comment saying we
opportunistically heal V0 rows on commit, or explicit `Set(...)` calls to match
Java.
##########
catalog/sql/sql.go:
##########
@@ -247,7 +247,76 @@ 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) migrateV0Schema(ctx context.Context) error {
+ hasCol, probeErr := c.icebergTypeColumnExists(ctx)
+ if probeErr != nil {
+ log.Printf("WARNING: skipping V0 schema migration; iceberg_type
column probe failed: %v", probeErr)
+
+ return nil
+ }
+ if hasCol {
+ return nil
+ }
+
+ if _, err := c.db.ExecContext(ctx, "ALTER TABLE iceberg_tables ADD
COLUMN iceberg_type VARCHAR(5)"); err != nil {
+ has, reprobeErr := c.icebergTypeColumnExists(ctx)
+ if has {
+ return nil
+ }
+ if reprobeErr != nil {
+ log.Printf("WARNING: V0 schema migration re-probe
failed after ALTER error: %v", reprobeErr)
+ }
+
+ return fmt.Errorf("migration of V0 schema failed: %w", err)
+ }
+
+ return nil
+}
+
+func (c *Catalog) icebergTypeColumnExists(ctx context.Context) (bool, error) {
+ var query string
+ switch dialectName := c.db.Dialect().Name().String(); dialectName {
+ case "pg":
+ query = `SELECT 1 FROM information_schema.columns
+ WHERE table_name = 'iceberg_tables'
+ AND column_name = 'iceberg_type' LIMIT 1`
+ case "mysql":
+ query = `SELECT 1 FROM information_schema.columns
+ WHERE table_schema = DATABASE()
+ AND table_name = 'iceberg_tables'
+ AND column_name = 'iceberg_type' LIMIT 1`
+ case "mssql":
+ query = `SELECT TOP 1 1 FROM information_schema.columns
+ WHERE table_name = 'iceberg_tables'
+ AND column_name = 'iceberg_type'`
+ case "sqlite":
+ query = `SELECT 1 FROM pragma_table_info('iceberg_tables')
+ WHERE name = 'iceberg_type' LIMIT 1`
+ case "oracle":
+ query = `SELECT 1 FROM user_tab_columns
+ WHERE UPPER(table_name) = 'ICEBERG_TABLES'
+ AND UPPER(column_name) = 'ICEBERG_TYPE'`
+ default:
+ return false, fmt.Errorf("unsupported dialect for V0 migration:
%s", dialectName)
+ }
+ var dummy int
+ if err := c.db.QueryRowContext(ctx, query).Scan(&dummy); err != nil {
+ if errors.Is(err, sql.ErrNoRows) {
+ return false, nil
+ }
+
+ return false, err
Review Comment:
Bare `err` here — the rest of the file wraps with `fmt.Errorf("...: %w",
err)`. Worth wrapping so the caller (and the warning log) gets some context
about what failed.
##########
catalog/sql/sql_test.go:
##########
@@ -322,6 +322,182 @@ 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.loadCatalogForTableCreation()
+
+ var present int
+ err = sqldb.QueryRow(
+ `SELECT 1 FROM pragma_table_info('iceberg_tables') WHERE name =
'iceberg_type'`,
+ ).Scan(&present)
+ s.Require().NoError(err, "migration must add iceberg_type column")
+ s.Equal(1, present)
+
+ var got sql.NullString
+ err = 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(err)
+ s.False(got.Valid, "pre-V0 row should keep iceberg_type IS NULL after
migration")
+
+ _ = s.loadCatalogForTableCreation()
+}
Review Comment:
This second `loadCatalogForTableCreation()` looks like it's meant to check
the migration is idempotent on re-open, but with no assertion or comment after
it, it isn't really pinning anything down. A line of comment plus a re-check
(column still present, row still NULL) would make the intent stick.
##########
catalog/sql/sql.go:
##########
@@ -247,7 +247,76 @@ 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) migrateV0Schema(ctx context.Context) error {
+ hasCol, probeErr := c.icebergTypeColumnExists(ctx)
+ if probeErr != nil {
+ log.Printf("WARNING: skipping V0 schema migration; iceberg_type
column probe failed: %v", probeErr)
+
+ return nil
+ }
+ if hasCol {
+ return nil
+ }
+
+ if _, err := c.db.ExecContext(ctx, "ALTER TABLE iceberg_tables ADD
COLUMN iceberg_type VARCHAR(5)"); err != nil {
+ has, reprobeErr := c.icebergTypeColumnExists(ctx)
+ if has {
+ return nil
+ }
+ if reprobeErr != nil {
+ log.Printf("WARNING: V0 schema migration re-probe
failed after ALTER error: %v", reprobeErr)
+ }
+
+ return fmt.Errorf("migration of V0 schema failed: %w", err)
+ }
+
+ return nil
+}
+
+func (c *Catalog) icebergTypeColumnExists(ctx context.Context) (bool, error) {
+ var query string
+ switch dialectName := c.db.Dialect().Name().String(); dialectName {
+ case "pg":
+ query = `SELECT 1 FROM information_schema.columns
+ WHERE table_name = 'iceberg_tables'
+ AND column_name = 'iceberg_type' LIMIT 1`
+ case "mysql":
+ query = `SELECT 1 FROM information_schema.columns
+ WHERE table_schema = DATABASE()
+ AND table_name = 'iceberg_tables'
+ AND column_name = 'iceberg_type' LIMIT 1`
+ case "mssql":
+ query = `SELECT TOP 1 1 FROM information_schema.columns
+ WHERE table_name = 'iceberg_tables'
+ AND column_name = 'iceberg_type'`
+ case "sqlite":
+ query = `SELECT 1 FROM pragma_table_info('iceberg_tables')
+ WHERE name = 'iceberg_type' LIMIT 1`
+ case "oracle":
+ query = `SELECT 1 FROM user_tab_columns
Review Comment:
Nit: this query's indentation mixes tabs and spaces where the other cases
use tabs. It's inside a raw string so gofmt won't catch it and the SQL runs
fine, but it'll look off in future diffs — worth realigning to match the others.
##########
catalog/sql/sql_test.go:
##########
@@ -322,6 +322,182 @@ 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.loadCatalogForTableCreation()
+
+ var present int
+ err = sqldb.QueryRow(
+ `SELECT 1 FROM pragma_table_info('iceberg_tables') WHERE name =
'iceberg_type'`,
+ ).Scan(&present)
+ s.Require().NoError(err, "migration must add iceberg_type column")
+ s.Equal(1, present)
+
+ var got sql.NullString
+ err = 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(err)
+ s.False(got.Valid, "pre-V0 row should keep iceberg_type IS NULL after
migration")
+
+ _ = s.loadCatalogForTableCreation()
+}
+
+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)
+
+ cat := s.loadCatalogForTableCreation()
+ 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()
Review Comment:
This exercises a row created at V1 and then manually NULLed — a fine proxy,
but not the same as a genuine V0 row where the column was absent until the
migration ran; a one-line comment noting the setup depends on migration having
already happened would help. Couple of smaller things in here too: the closure
hardcodes `"default"` as the catalog name (`cat.Name()` would be less fragile),
and `EqualValues(1, n)` could be `Equal(int64(1), n)` for a stricter check.
##########
catalog/sql/sql_test.go:
##########
@@ -322,6 +322,182 @@ func (s *SqliteCatalogTestSuite)
TestCreationAllTablesExist() {
s.confirmTablesExist(sqldb)
}
+func (s *SqliteCatalogTestSuite) TestV0SchemaFromIcebergIsMigrated() {
+ sqldb := s.getDB()
Review Comment:
The coverage here is all SQLite, which leaves the riskiest paths untested:
the `mssql`/`oracle` probe + ALTER (where the `ADD COLUMN` syntax bug lives),
the probe-error → skip-migration branch, and the concurrent-ALTER re-probe
recovery. The first in particular means CI won't catch a V0 catalog being
unopenable on those engines. Even a focused test that injects a failing probe
would cover the swallow path.
##########
catalog/sql/sql.go:
##########
@@ -247,7 +247,76 @@ 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) migrateV0Schema(ctx context.Context) error {
+ hasCol, probeErr := c.icebergTypeColumnExists(ctx)
+ if probeErr != nil {
+ log.Printf("WARNING: skipping V0 schema migration; iceberg_type
column probe failed: %v", probeErr)
+
+ return nil
+ }
+ if hasCol {
+ return nil
+ }
+
+ if _, err := c.db.ExecContext(ctx, "ALTER TABLE iceberg_tables ADD
COLUMN iceberg_type VARCHAR(5)"); err != nil {
Review Comment:
This ALTER is hardcoded, but `ADD COLUMN` isn't valid on every dialect the
probe handles. SQL Server wants `ALTER TABLE iceberg_tables ADD iceberg_type
VARCHAR(5)` (no `COLUMN`) and Oracle wants `ADD (iceberg_type VARCHAR2(5))`.
Since the probe already has `mssql` and `oracle` cases, a V0 catalog on either
engine reaches this line, the ALTER fails, the re-probe still finds the column
missing, and `migrateV0Schema` returns a hard error — so the catalog can't be
opened at all. I'd switch on the dialect here the same way the probe does and
emit dialect-specific DDL.
Separately, `VARCHAR(5)` diverges from the unbounded `VARCHAR` bun emits for
fresh Go-created tables, so migrated and fresh DBs end up with different column
types — and it fits `"TABLE"` exactly, which is tight. I'd drop the length
bound to match what bun generates.
--
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]
