This is an automated email from the ASF dual-hosted git repository.
zeroshade pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg-go.git
The following commit(s) were added to refs/heads/main by this push:
new 60abc85 feat(catalog/sql): Add views related ops (#414)
60abc85 is described below
commit 60abc852c842859a2597e62c20f11d9f60c7fe1d
Author: Dao Thanh Tung <[email protected]>
AuthorDate: Fri Jun 20 16:14:29 2025 +0100
feat(catalog/sql): Add views related ops (#414)
Signed-off-by: dttung2905 <[email protected]>
---
README.md | 9 +-
catalog/internal/utils.go | 158 ++++++++++++++++++++++++++++++
catalog/sql/sql.go | 241 +++++++++++++++++++++++++++++++++++++++++++++-
catalog/sql/sql_test.go | 144 ++++++++++++++++++++++++++-
4 files changed, 544 insertions(+), 8 deletions(-)
diff --git a/README.md b/README.md
index 63d277c..a699e4d 100644
--- a/README.md
+++ b/README.md
@@ -78,10 +78,11 @@ $ cd iceberg-go/cmd/iceberg && go build .
| Check Namespace Exists | X | | | X | X |
| Drop Namespace | X | | | X | X |
| Update Namespace Properties | X | | | X | X |
-| Create View | X | | | | |
-| List View | X | | | | |
-| Drop View | X | | | | |
-| Check View Exists | X | | | | |
+| Create View | X | | | | X |
+| Load View | | | | | X |
+| List View | X | | | | X |
+| Drop View | X | | | | X |
+| Check View Exists | X | | | | X |
### Read/Write Data Support
diff --git a/catalog/internal/utils.go b/catalog/internal/utils.go
index 81b8cc3..563756a 100644
--- a/catalog/internal/utils.go
+++ b/catalog/internal/utils.go
@@ -28,6 +28,7 @@ import (
"regexp"
"strconv"
"strings"
+ "time"
"github.com/apache/iceberg-go"
"github.com/apache/iceberg-go/catalog"
@@ -251,3 +252,160 @@ func UpdateAndStageTable(ctx context.Context, current
*table.Table, ident table.
),
}, nil
}
+
+func CreateViewMetadata(
+ ctx context.Context,
+ catalogName string,
+ nsIdent []string,
+ schema *iceberg.Schema,
+ viewSQL string,
+ loc string,
+ props iceberg.Properties,
+) (metadataLocation string, err error) {
+ versionId := int64(1)
+ timestampMs := time.Now().UnixMilli()
+
+ viewVersion := struct {
+ VersionID int64 `json:"version-id"`
+ TimestampMs int64 `json:"timestamp-ms"`
+ SchemaID int `json:"schema-id"`
+ Summary map[string]string `json:"summary"`
+ Operation string `json:"operation"`
+ Representations []struct {
+ Type string `json:"type"`
+ SQL string `json:"sql"`
+ Dialect string `json:"dialect"`
+ } `json:"representations"`
+ DefaultCatalog string `json:"default-catalog"`
+ DefaultNamespace []string `json:"default-namespace"`
+ }{
+ VersionID: versionId,
+ TimestampMs: timestampMs,
+ SchemaID: schema.ID,
+ Summary: map[string]string{"sql": viewSQL},
+ Operation: "create",
+ Representations: []struct {
+ Type string `json:"type"`
+ SQL string `json:"sql"`
+ Dialect string `json:"dialect"`
+ }{
+ {Type: "sql", SQL: viewSQL, Dialect: "default"},
+ },
+ DefaultCatalog: catalogName,
+ DefaultNamespace: nsIdent,
+ }
+
+ viewVersionBytes, err := json.Marshal(viewVersion)
+ if err != nil {
+ return "", fmt.Errorf("failed to marshal view version: %w", err)
+ }
+
+ if props == nil {
+ props = iceberg.Properties{}
+ }
+ props["view-version"] = string(viewVersionBytes)
+ props["view-format"] = "iceberg"
+ props["view-sql"] = viewSQL
+
+ metadataLocation = loc + "/metadata/view-" + uuid.New().String() +
".metadata.json"
+
+ viewUUID := uuid.New().String()
+ props["view-uuid"] = viewUUID
+
+ viewMetadata := map[string]interface{}{
+ "view-uuid": viewUUID,
+ "format-version": 1,
+ "location": loc,
+ "schema": schema,
+ "current-version-id": versionId,
+ "versions": map[string]interface{}{
+ "1": viewVersion,
+ },
+ "properties": props,
+ "version-log": []map[string]interface{}{
+ {
+ "timestamp-ms": timestampMs,
+ "version-id": versionId,
+ },
+ },
+ }
+
+ viewMetadataBytes, err := json.Marshal(viewMetadata)
+ if err != nil {
+ return "", fmt.Errorf("failed to marshal view metadata: %w",
err)
+ }
+
+ fs, err := io.LoadFS(ctx, props, metadataLocation)
+ if err != nil {
+ return "", fmt.Errorf("failed to load filesystem for view
metadata: %w", err)
+ }
+
+ wfs, ok := fs.(io.WriteFileIO)
+ if !ok {
+ return "", errors.New("filesystem IO does not support writing")
+ }
+
+ out, err := wfs.Create(metadataLocation)
+ if err != nil {
+ return "", fmt.Errorf("failed to create view metadata file:
%w", err)
+ }
+ defer out.Close()
+
+ if _, err := out.Write(viewMetadataBytes); err != nil {
+ return "", fmt.Errorf("failed to write view metadata: %w", err)
+ }
+
+ return metadataLocation, nil
+}
+
+func LoadViewMetadata(ctx context.Context,
+ props iceberg.Properties,
+ metadataLocation string,
+ viewName string,
+ namespace string,
+) (map[string]interface{}, error) {
+ // Initial metadata with basic information
+ viewMetadata := map[string]interface{}{
+ "name": viewName,
+ "namespace": namespace,
+ "metadata-location": metadataLocation,
+ }
+
+ // Load the filesystem
+ fs, err := io.LoadFS(ctx, props, metadataLocation)
+ if err != nil {
+ return nil, fmt.Errorf("error loading view metadata: %w", err)
+ }
+
+ // Open the metadata file
+ inputFile, err := fs.Open(metadataLocation)
+ if err != nil {
+ return viewMetadata, fmt.Errorf("error encountered loading view
metadata: %w", err)
+ }
+ defer inputFile.Close()
+
+ // Decode the complete metadata
+ var fullViewMetadata map[string]interface{}
+ if err := json.NewDecoder(inputFile).Decode(&fullViewMetadata); err !=
nil {
+ return viewMetadata, fmt.Errorf("error encountered decoding
view metadata: %w", err)
+ }
+
+ // Update the metadata with name, namespace and location
+ fullViewMetadata["name"] = viewName
+ fullViewMetadata["namespace"] = namespace
+ fullViewMetadata["metadata-location"] = metadataLocation
+
+ if props, ok :=
fullViewMetadata["properties"].(map[string]interface{}); ok {
+ strProps := make(map[string]string)
+ for k, v := range props {
+ if str, ok := v.(string); ok {
+ strProps[k] = str
+ } else if vJson, err := json.Marshal(v); err == nil {
+ strProps[k] = string(vJson)
+ }
+ }
+ fullViewMetadata["properties"] = strProps
+ }
+
+ return fullViewMetadata, nil
+}
diff --git a/catalog/sql/sql.go b/catalog/sql/sql.go
index 762a643..8f2f29f 100644
--- a/catalog/sql/sql.go
+++ b/catalog/sql/sql.go
@@ -61,6 +61,11 @@ const (
initCatalogTablesKey = "init_catalog_tables"
)
+const (
+ TableType = "TABLE"
+ ViewType = "VIEW"
+)
+
func init() {
catalog.Register("sql", catalog.RegistrarFunc(func(ctx context.Context,
name string, p iceberg.Properties) (c catalog.Catalog, err error) {
driver, ok := p[DriverKey]
@@ -133,6 +138,7 @@ type sqlIcebergTable struct {
CatalogName string `bun:",pk"`
TableNamespace string `bun:",pk"`
TableName string `bun:",pk"`
+ IcebergType string // TableType or ViewType
MetadataLocation sql.NullString
PreviousMetadataLocation sql.NullString
}
@@ -302,6 +308,7 @@ func (c *Catalog) CreateTable(ctx context.Context, ident
table.Identifier, sc *i
TableNamespace: ns,
TableName: tblIdent,
MetadataLocation: sql.NullString{String:
staged.MetadataLocation(), Valid: true},
+ IcebergType: TableType,
}).Exec(ctx)
if err != nil {
return fmt.Errorf("failed to create table: %w", err)
@@ -345,9 +352,11 @@ func (c *Catalog) CommitTable(ctx context.Context, tbl
*table.Table, reqs []tabl
CatalogName: c.name,
TableNamespace: strings.Join(ns, "."),
TableName: tblName,
+ IcebergType: TableType,
MetadataLocation: sql.NullString{Valid:
true, String: staged.MetadataLocation()},
PreviousMetadataLocation: sql.NullString{Valid:
true, String: current.MetadataLocation()},
}).WherePK().Where("metadata_location = ?",
current.MetadataLocation()).
+ Where("iceberg_type = ?", TableType).
Exec(ctx)
if err != nil {
return fmt.Errorf("error updating table
information: %w", err)
@@ -369,6 +378,7 @@ func (c *Catalog) CommitTable(ctx context.Context, tbl
*table.Table, reqs []tabl
CatalogName: c.name,
TableNamespace: strings.Join(ns, "."),
TableName: tblName,
+ IcebergType: TableType,
MetadataLocation: sql.NullString{Valid: true, String:
staged.MetadataLocation()},
}).Exec(ctx)
if err != nil {
@@ -438,7 +448,7 @@ func (c *Catalog) DropTable(ctx context.Context, identifier
table.Identifier) er
CatalogName: c.name,
TableNamespace: ns,
TableName: tbl,
- }).WherePK().Exec(ctx)
+ }).WherePK().Where("iceberg_type = ?", TableType).Exec(ctx)
if err != nil {
return fmt.Errorf("failed to delete table entry: %w",
err)
}
@@ -489,7 +499,7 @@ func (c *Catalog) RenameTable(ctx context.Context, from, to
table.Identifier) (*
CatalogName: c.name,
TableNamespace: fromNs,
TableName: fromTbl,
- }).WherePK().
+ }).WherePK().Where("iceberg_type = ?", TableType).
Set("table_namespace = ?", toNs).
Set("table_name = ?", toTbl).
Exec(ctx)
@@ -679,6 +689,7 @@ func (c *Catalog) listTablesAll(ctx context.Context,
namespace table.Identifier)
err := tx.NewSelect().Model(&tables).
Where("catalog_name = ?", c.name).
Where("table_namespace = ?", ns).
+ Where("iceberg_type = ?", TableType).
Scan(ctx)
return tables, err
@@ -814,3 +825,229 @@ func (c *Catalog) UpdateNamespaceProperties(ctx
context.Context, namespace table
func (c *Catalog) CheckNamespaceExists(ctx context.Context, namespace
table.Identifier) (bool, error) {
return c.namespaceExists(ctx, strings.Join(namespace, "."))
}
+
+// CreateView creates a new view in the catalog.
+func (c *Catalog) CreateView(ctx context.Context, identifier table.Identifier,
schema *iceberg.Schema, viewSQL string, props iceberg.Properties) error {
+ nsIdent := catalog.NamespaceFromIdent(identifier)
+ viewIdent := catalog.TableNameFromIdent(identifier)
+ ns := strings.Join(nsIdent, ".")
+
+ exists, err := c.namespaceExists(ctx, ns)
+ if err != nil {
+ return err
+ }
+ if !exists {
+ return fmt.Errorf("%w: %s", catalog.ErrNoSuchNamespace, ns)
+ }
+
+ exists, err = c.CheckViewExists(ctx, identifier)
+ if err != nil {
+ return err
+ }
+ if exists {
+ return fmt.Errorf("%w: %s", catalog.ErrViewAlreadyExists,
identifier)
+ }
+
+ loc, err := internal.ResolveTableLocation(ctx, "", ns, viewIdent,
c.props, c.LoadNamespaceProperties)
+ if err != nil {
+ return err
+ }
+
+ metadataLocation, err := internal.CreateViewMetadata(ctx, c.name,
nsIdent, schema, viewSQL, loc, props)
+ if err != nil {
+ return err
+ }
+ err = withWriteTx(ctx, c.db, func(ctx context.Context, tx bun.Tx) error
{
+ _, err := tx.NewInsert().Model(&sqlIcebergTable{
+ CatalogName: c.name,
+ TableNamespace: ns,
+ TableName: viewIdent,
+ IcebergType: ViewType,
+ MetadataLocation: sql.NullString{String:
metadataLocation, Valid: true},
+ }).Exec(ctx)
+ if err != nil {
+ return fmt.Errorf("failed to create view: %w", err)
+ }
+
+ return nil
+ })
+
+ return err
+}
+
+// ListViews returns a list of view identifiers in the catalog.
+func (c *Catalog) ListViews(ctx context.Context, namespace table.Identifier)
iter.Seq2[table.Identifier, error] {
+ views, err := c.listViewsAll(ctx, namespace)
+ if err != nil {
+ return func(yield func(table.Identifier, error) bool) {
+ yield(table.Identifier{}, err)
+ }
+ }
+
+ return func(yield func(table.Identifier, error) bool) {
+ for _, v := range views {
+ if !yield(v, nil) {
+ return
+ }
+ }
+ }
+}
+
+func (c *Catalog) listViewsAll(ctx context.Context, namespace
table.Identifier) ([]table.Identifier, error) {
+ if len(namespace) > 0 {
+ exists, err := c.namespaceExists(ctx, strings.Join(namespace,
"."))
+ if err != nil {
+ return nil, err
+ }
+ if !exists {
+ return nil, fmt.Errorf("%w: %s",
catalog.ErrNoSuchNamespace, strings.Join(namespace, "."))
+ }
+ }
+
+ ns := strings.Join(namespace, ".")
+ views, err := withReadTx(ctx, c.db, func(ctx context.Context, tx
bun.Tx) ([]sqlIcebergTable, error) {
+ var views []sqlIcebergTable
+ err := tx.NewSelect().Model(&views).
+ Where("catalog_name = ?", c.name).
+ Where("table_namespace = ?", ns).
+ Where("iceberg_type = ?", ViewType).
+ Scan(ctx)
+
+ return views, err
+ })
+ if err != nil {
+ return nil, fmt.Errorf("error listing views for namespace '%s':
%w", namespace, err)
+ }
+
+ ret := make([]table.Identifier, len(views))
+ for i, v := range views {
+ ret[i] = append(strings.Split(v.TableNamespace, "."),
v.TableName)
+ }
+
+ return ret, nil
+}
+
+// DropView deletes a view from the catalog.
+func (c *Catalog) DropView(ctx context.Context, identifier table.Identifier)
error {
+ ns := strings.Join(catalog.NamespaceFromIdent(identifier), ".")
+ viewName := catalog.TableNameFromIdent(identifier)
+
+ metadataLocation := ""
+ view, err := withReadTx(ctx, c.db, func(ctx context.Context, tx bun.Tx)
(*sqlIcebergTable, error) {
+ v := new(sqlIcebergTable)
+ err := tx.NewSelect().Model(v).
+ Where("catalog_name = ?", c.name).
+ Where("table_namespace = ?", ns).
+ Where("table_name = ?", viewName).
+ Where("iceberg_type = ?", ViewType).
+ Scan(ctx)
+ if errors.Is(err, sql.ErrNoRows) {
+ return nil, fmt.Errorf("%w: %s", catalog.ErrNoSuchView,
identifier)
+ }
+ if err != nil {
+ return nil, fmt.Errorf("error encountered loading view
%s: %w", identifier, err)
+ }
+
+ return v, nil
+ })
+ if err != nil {
+ return err
+ }
+
+ if view.MetadataLocation.Valid {
+ metadataLocation = view.MetadataLocation.String
+ }
+
+ err = withWriteTx(ctx, c.db, func(ctx context.Context, tx bun.Tx) error
{
+ res, err := tx.NewDelete().Model(&sqlIcebergTable{
+ CatalogName: c.name,
+ TableNamespace: ns,
+ TableName: viewName,
+ }).WherePK().Where("iceberg_type = ?", ViewType).Exec(ctx)
+ if err != nil {
+ return fmt.Errorf("failed to delete view entry: %w",
err)
+ }
+
+ n, err := res.RowsAffected()
+ if err != nil {
+ return fmt.Errorf("error encountered when deleting view
entry: %w", err)
+ }
+
+ if n == 0 {
+ return fmt.Errorf("%w: %s", catalog.ErrNoSuchView,
identifier)
+ }
+
+ return nil
+ })
+ if err != nil {
+ return err
+ }
+
+ if metadataLocation != "" {
+ fs, err := io.LoadFS(ctx, c.props, metadataLocation)
+ if err != nil {
+ return nil
+ }
+
+ _ = fs.Remove(metadataLocation)
+ }
+
+ return nil
+}
+
+// CheckViewExists returns true if a view exists in the catalog.
+func (c *Catalog) CheckViewExists(ctx context.Context, identifier
table.Identifier) (bool, error) {
+ ns := strings.Join(catalog.NamespaceFromIdent(identifier), ".")
+ viewName := catalog.TableNameFromIdent(identifier)
+
+ return withReadTx(ctx, c.db, func(ctx context.Context, tx bun.Tx)
(bool, error) {
+ exists, err := tx.NewSelect().Model(&sqlIcebergTable{
+ CatalogName: c.name,
+ TableNamespace: ns,
+ TableName: viewName,
+ }).WherePK().Where("iceberg_type = ?", ViewType).Exists(ctx)
+ if err != nil {
+ return false, fmt.Errorf("error checking view
existence: %w", err)
+ }
+
+ return exists, nil
+ })
+}
+
+// LoadView loads a view from the catalog.
+func (c *Catalog) LoadView(ctx context.Context, identifier table.Identifier)
(map[string]interface{}, error) {
+ ns := strings.Join(catalog.NamespaceFromIdent(identifier), ".")
+ viewName := catalog.TableNameFromIdent(identifier)
+
+ view, err := withReadTx(ctx, c.db, func(ctx context.Context, tx bun.Tx)
(*sqlIcebergTable, error) {
+ v := new(sqlIcebergTable)
+ err := tx.NewSelect().Model(v).
+ Where("catalog_name = ?", c.name).
+ Where("table_namespace = ?", ns).
+ Where("table_name = ?", viewName).
+ Where("iceberg_type = ?", ViewType).
+ Scan(ctx)
+ if errors.Is(err, sql.ErrNoRows) {
+ return nil, fmt.Errorf("%w: %s", catalog.ErrNoSuchView,
identifier)
+ }
+ if err != nil {
+ return nil, fmt.Errorf("error encountered loading view
%s: %w", identifier, err)
+ }
+
+ return v, nil
+ })
+ if err != nil {
+ return nil, err
+ }
+
+ if !view.MetadataLocation.Valid {
+ return nil, fmt.Errorf("%w: %s, metadata location is missing",
catalog.ErrNoSuchView, identifier)
+ }
+
+ viewMetadata, err := internal.LoadViewMetadata(ctx, c.props,
view.MetadataLocation.String, viewName, ns)
+ if err != nil {
+ return nil, err
+ }
+
+ return viewMetadata, nil
+}
diff --git a/catalog/sql/sql_test.go b/catalog/sql/sql_test.go
index 496f230..b6fd589 100644
--- a/catalog/sql/sql_test.go
+++ b/catalog/sql/sql_test.go
@@ -280,7 +280,8 @@ func (s *SqliteCatalogTestSuite)
TestCreationOneTableExists() {
_, err := sqldb.Exec(`CREATE TABLE "iceberg_tables" (
"catalog_name" VARCHAR NOT NULL,
"table_namespace" VARCHAR NOT NULL,
- "table_name" VARCHAR NOT NULL,
+ "table_name" VARCHAR NOT NULL,
+ "iceberg_type" VARCHAR NOT NULL DEFAULT 'TABLE',
"metadata_location" VARCHAR,
"previous_metadata_location" VARCHAR,
PRIMARY KEY ("catalog_name", "table_namespace", "table_name"))`)
@@ -298,7 +299,8 @@ func (s *SqliteCatalogTestSuite)
TestCreationAllTablesExist() {
_, err := sqldb.Exec(`CREATE TABLE "iceberg_tables" (
"catalog_name" VARCHAR NOT NULL,
"table_namespace" VARCHAR NOT NULL,
- "table_name" VARCHAR NOT NULL,
+ "table_name" VARCHAR NOT NULL,
+ "iceberg_type" VARCHAR,
"metadata_location" VARCHAR,
"previous_metadata_location" VARCHAR,
PRIMARY KEY ("catalog_name", "table_namespace", "table_name"))`)
@@ -1041,6 +1043,144 @@ func (s *SqliteCatalogTestSuite) TestCommitTable() {
}
}
+func (s *SqliteCatalogTestSuite) TestCreateView() {
+ db := s.getCatalogSqlite()
+ s.Require().NoError(db.CreateSQLTables(context.Background()))
+
+ nsName := databaseName()
+ viewName := tableName()
+ s.Require().NoError(db.CreateNamespace(context.Background(),
[]string{nsName}, nil))
+
+ viewSQL := "SELECT * FROM test_table"
+ schema := iceberg.NewSchema(1, iceberg.NestedField{
+ ID: 1, Name: "id", Type: iceberg.PrimitiveTypes.Int32,
Required: true,
+ })
+ s.Require().NoError(db.CreateView(context.Background(),
[]string{nsName, viewName}, schema, viewSQL, nil))
+
+ exists, err := db.CheckViewExists(context.Background(),
[]string{nsName, viewName})
+ s.Require().NoError(err)
+ s.True(exists)
+}
+
+func (s *SqliteCatalogTestSuite) TestDropView() {
+ db := s.getCatalogSqlite()
+ s.Require().NoError(db.CreateSQLTables(context.Background()))
+
+ nsName := databaseName()
+ viewName := tableName()
+ s.Require().NoError(db.CreateNamespace(context.Background(),
[]string{nsName}, nil))
+
+ viewSQL := "SELECT * FROM test_table"
+ schema := iceberg.NewSchema(1, iceberg.NestedField{
+ ID: 1, Name: "id", Type: iceberg.PrimitiveTypes.Int32,
Required: true,
+ })
+ s.Require().NoError(db.CreateView(context.Background(),
[]string{nsName, viewName}, schema, viewSQL, nil))
+
+ exists, err := db.CheckViewExists(context.Background(),
[]string{nsName, viewName})
+ s.Require().NoError(err)
+ s.True(exists)
+
+ s.Require().NoError(db.DropView(context.Background(), []string{nsName,
viewName}))
+
+ exists, err = db.CheckViewExists(context.Background(), []string{nsName,
viewName})
+ s.Require().NoError(err)
+ s.False(exists)
+
+ err = db.DropView(context.Background(), []string{nsName, "nonexistent"})
+ s.Error(err)
+ s.ErrorIs(err, catalog.ErrNoSuchView)
+}
+
+func (s *SqliteCatalogTestSuite) TestCheckViewExists() {
+ db := s.getCatalogSqlite()
+ s.Require().NoError(db.CreateSQLTables(context.Background()))
+
+ nsName := databaseName()
+ viewName := tableName()
+ s.Require().NoError(db.CreateNamespace(context.Background(),
[]string{nsName}, nil))
+
+ exists, err := db.CheckViewExists(context.Background(),
[]string{nsName, viewName})
+ s.Require().NoError(err)
+ s.False(exists)
+
+ viewSQL := "SELECT * FROM test_table"
+ schema := iceberg.NewSchema(1, iceberg.NestedField{
+ ID: 1, Name: "id", Type: iceberg.PrimitiveTypes.Int32,
Required: true,
+ })
+ s.Require().NoError(db.CreateView(context.Background(),
[]string{nsName, viewName}, schema, viewSQL, nil))
+
+ exists, err = db.CheckViewExists(context.Background(), []string{nsName,
viewName})
+ s.Require().NoError(err)
+ s.True(exists)
+}
+
+func (s *SqliteCatalogTestSuite) TestListViews() {
+ db := s.getCatalogSqlite()
+ s.Require().NoError(db.CreateSQLTables(context.Background()))
+
+ nsName := databaseName()
+ s.Require().NoError(db.CreateNamespace(context.Background(),
[]string{nsName}, nil))
+
+ viewNames := []string{tableName(), tableName(), tableName()}
+ for _, viewName := range viewNames {
+ viewSQL := "SELECT * FROM test_table"
+ schema := iceberg.NewSchema(1, iceberg.NestedField{
+ ID: 1, Name: "id", Type: iceberg.PrimitiveTypes.Int32,
Required: true,
+ })
+ s.Require().NoError(db.CreateView(context.Background(),
[]string{nsName, viewName}, schema, viewSQL, nil))
+ }
+
+ var foundViews []table.Identifier
+ viewsIter := db.ListViews(context.Background(), []string{nsName})
+ for view, err := range viewsIter {
+ s.Require().NoError(err)
+ foundViews = append(foundViews, view)
+ }
+
+ s.Require().Len(foundViews, len(viewNames))
+ for _, view := range foundViews {
+ s.Equal(nsName, view[0])
+ s.Contains(viewNames, view[1])
+ }
+
+ viewsIter = db.ListViews(context.Background(), []string{"nonexistent"})
+ for _, err := range viewsIter {
+ s.Error(err)
+ s.ErrorIs(err, catalog.ErrNoSuchNamespace)
+
+ break
+ }
+}
+
+func (s *SqliteCatalogTestSuite) TestLoadView() {
+ db := s.getCatalogSqlite()
+ s.Require().NoError(db.CreateSQLTables(context.Background()))
+
+ nsName := databaseName()
+ viewName := tableName()
+ s.Require().NoError(db.CreateNamespace(context.Background(),
[]string{nsName}, nil))
+
+ viewSQL := "SELECT * FROM test_table"
+ schema := iceberg.NewSchema(1, iceberg.NestedField{
+ ID: 1, Name: "id", Type: iceberg.PrimitiveTypes.Int32,
Required: true,
+ })
+ props := iceberg.Properties{
+ "comment": "Test view",
+ "owner": "test-user",
+ }
+ s.Require().NoError(db.CreateView(context.Background(),
[]string{nsName, viewName}, schema, viewSQL, props))
+
+ viewInfo, err := db.LoadView(context.Background(), []string{nsName,
viewName})
+ s.Require().NoError(err)
+
+ s.Equal(viewName, viewInfo["name"])
+ s.Equal(nsName, viewInfo["namespace"])
+
+ _, err = db.LoadView(context.Background(), []string{nsName,
"nonexistent"})
+ s.Error(err)
+ s.ErrorIs(err, catalog.ErrNoSuchView)
+}
+
func TestSqlCatalog(t *testing.T) {
suite.Run(t, new(SqliteCatalogTestSuite))
}
