From 368fb9d2676c3d8ae0ed80305d79469652b3eed5 Mon Sep 17 00:00:00 2001 From: Kevin K Biju <52661649+heavycrystal@users.noreply.github.com> Date: Mon, 6 Nov 2023 18:59:41 +0000 Subject: [PATCH] added metadata_schema option for SF and PG peers (#560) All metadata tables and Snowflake stages for CDC and QRep mirrors will be created in the provided schema, schema will be auto-created. Default is `_peerdb_internal` fixes: https://github.com/PeerDB-io/peerdb/issues/439 --- .gitignore | 3 +- flow/connectors/postgres/client.go | 31 +- flow/connectors/postgres/postgres.go | 32 +- flow/connectors/postgres/qrep.go | 48 ++- flow/connectors/postgres/qrep_sync_method.go | 7 +- flow/connectors/snowflake/qrep.go | 42 ++- flow/connectors/snowflake/snowflake.go | 63 ++-- flow/generated/protos/peers.pb.go | 369 ++++++++++--------- nexus/analyzer/src/lib.rs | 2 + nexus/catalog/src/lib.rs | 1 + nexus/pt/src/peerdb_peers.rs | 6 + nexus/pt/src/peerdb_peers.serde.rs | 36 ++ protos/peers.proto | 5 +- ui/grpc_generated/peers.ts | 49 ++- 14 files changed, 439 insertions(+), 255 deletions(-) diff --git a/.gitignore b/.gitignore index 8df46502df..07ce752e93 100644 --- a/.gitignore +++ b/.gitignore @@ -1,4 +1,5 @@ -.vscode +.vscode/ .env tmp/ .envrc +.idea/ diff --git a/flow/connectors/postgres/client.go b/flow/connectors/postgres/client.go index c9bed66d2f..ad641cf01a 100644 --- a/flow/connectors/postgres/client.go +++ b/flow/connectors/postgres/client.go @@ -17,13 +17,12 @@ import ( //nolint:stylecheck const ( - internalSchema = "_peerdb_internal" mirrorJobsTableIdentifier = "peerdb_mirror_jobs" createMirrorJobsTableSQL = `CREATE TABLE IF NOT EXISTS %s.%s(mirror_job_name TEXT PRIMARY KEY, lsn_offset BIGINT NOT NULL,sync_batch_id BIGINT NOT NULL,normalize_batch_id BIGINT NOT NULL)` - rawTablePrefix = "_peerdb_raw" - createInternalSchemaSQL = "CREATE SCHEMA IF NOT EXISTS %s" - createRawTableSQL = `CREATE TABLE IF NOT EXISTS %s.%s(_peerdb_uid TEXT NOT NULL, + rawTablePrefix = "_peerdb_raw" + createSchemaSQL = "CREATE SCHEMA IF NOT EXISTS %s" + createRawTableSQL = `CREATE TABLE IF NOT EXISTS %s.%s(_peerdb_uid TEXT NOT NULL, _peerdb_timestamp BIGINT NOT NULL,_peerdb_destination_table_name TEXT NOT NULL,_peerdb_data JSONB NOT NULL, _peerdb_record_type INTEGER NOT NULL, _peerdb_match_data JSONB,_peerdb_batch_id INTEGER, _peerdb_unchanged_toast_columns TEXT)` @@ -292,8 +291,8 @@ func (c *PostgresConnector) createSlotAndPublication( return nil } -func (c *PostgresConnector) createInternalSchema(createSchemaTx pgx.Tx) error { - _, err := createSchemaTx.Exec(c.ctx, fmt.Sprintf(createInternalSchemaSQL, internalSchema)) +func (c *PostgresConnector) createMetadataSchema(createSchemaTx pgx.Tx) error { + _, err := createSchemaTx.Exec(c.ctx, fmt.Sprintf(createSchemaSQL, c.metadataSchema)) if err != nil { return fmt.Errorf("error while creating internal schema: %w", err) } @@ -329,7 +328,7 @@ func generateCreateTableSQLForNormalizedTable(sourceTableIdentifier string, func (c *PostgresConnector) GetLastSyncBatchID(jobName string) (int64, error) { rows, err := c.pool.Query(c.ctx, fmt.Sprintf( getLastSyncBatchID_SQL, - internalSchema, + c.metadataSchema, mirrorJobsTableIdentifier, ), jobName) if err != nil { @@ -350,7 +349,7 @@ func (c *PostgresConnector) GetLastSyncBatchID(jobName string) (int64, error) { } func (c *PostgresConnector) getLastNormalizeBatchID(jobName string) (int64, error) { - rows, err := c.pool.Query(c.ctx, fmt.Sprintf(getLastNormalizeBatchID_SQL, internalSchema, + rows, err := c.pool.Query(c.ctx, fmt.Sprintf(getLastNormalizeBatchID_SQL, c.metadataSchema, mirrorJobsTableIdentifier), jobName) if err != nil { return 0, fmt.Errorf("error querying Postgres peer for last normalizeBatchId: %w", err) @@ -371,7 +370,7 @@ func (c *PostgresConnector) getLastNormalizeBatchID(jobName string) (int64, erro func (c *PostgresConnector) jobMetadataExists(jobName string) (bool, error) { rows, err := c.pool.Query(c.ctx, - fmt.Sprintf(checkIfJobMetadataExistsSQL, internalSchema, mirrorJobsTableIdentifier), jobName) + fmt.Sprintf(checkIfJobMetadataExistsSQL, c.metadataSchema, mirrorJobsTableIdentifier), jobName) if err != nil { return false, fmt.Errorf("failed to check if job exists: %w", err) } @@ -405,14 +404,14 @@ func (c *PostgresConnector) updateSyncMetadata(flowJobName string, lastCP int64, if !jobMetadataExists { _, err := syncRecordsTx.Exec(c.ctx, - fmt.Sprintf(insertJobMetadataSQL, internalSchema, mirrorJobsTableIdentifier), + fmt.Sprintf(insertJobMetadataSQL, c.metadataSchema, mirrorJobsTableIdentifier), flowJobName, lastCP, syncBatchID, 0) if err != nil { return fmt.Errorf("failed to insert flow job status: %w", err) } } else { _, err := syncRecordsTx.Exec(c.ctx, - fmt.Sprintf(updateMetadataForSyncRecordsSQL, internalSchema, mirrorJobsTableIdentifier), + fmt.Sprintf(updateMetadataForSyncRecordsSQL, c.metadataSchema, mirrorJobsTableIdentifier), lastCP, syncBatchID, flowJobName) if err != nil { return fmt.Errorf("failed to update flow job status: %w", err) @@ -433,7 +432,7 @@ func (c *PostgresConnector) updateNormalizeMetadata(flowJobName string, normaliz } _, err = normalizeRecordsTx.Exec(c.ctx, - fmt.Sprintf(updateMetadataForNormalizeRecordsSQL, internalSchema, mirrorJobsTableIdentifier), + fmt.Sprintf(updateMetadataForNormalizeRecordsSQL, c.metadataSchema, mirrorJobsTableIdentifier), normalizeBatchID, flowJobName) if err != nil { return fmt.Errorf("failed to update metadata for NormalizeTables: %w", err) @@ -446,7 +445,7 @@ func (c *PostgresConnector) getTableNametoUnchangedCols(flowJobName string, sync normalizeBatchID int64) (map[string][]string, error) { rawTableIdentifier := getRawTableIdentifier(flowJobName) - rows, err := c.pool.Query(c.ctx, fmt.Sprintf(getTableNameToUnchangedToastColsSQL, internalSchema, + rows, err := c.pool.Query(c.ctx, fmt.Sprintf(getTableNameToUnchangedToastColsSQL, c.metadataSchema, rawTableIdentifier), normalizeBatchID, syncBatchID) if err != nil { return nil, fmt.Errorf("error while retrieving table names for normalization: %w", err) @@ -512,11 +511,11 @@ func (c *PostgresConnector) generateFallbackStatements(destinationTableIdentifie deleteWhereClauseSQL := strings.TrimSuffix(strings.Join(deleteWhereClauseArray, ""), "AND ") fallbackUpsertStatement := fmt.Sprintf(fallbackUpsertStatementSQL, - strings.TrimSuffix(strings.Join(maps.Values(primaryKeyColumnCasts), ","), ","), internalSchema, + strings.TrimSuffix(strings.Join(maps.Values(primaryKeyColumnCasts), ","), ","), c.metadataSchema, rawTableIdentifier, destinationTableIdentifier, insertColumnsSQL, flattenedCastsSQL, strings.Join(normalizedTableSchema.PrimaryKeyColumns, ","), updateColumnsSQL) fallbackDeleteStatement := fmt.Sprintf(fallbackDeleteStatementSQL, - strings.Join(maps.Values(primaryKeyColumnCasts), ","), internalSchema, + strings.Join(maps.Values(primaryKeyColumnCasts), ","), c.metadataSchema, rawTableIdentifier, destinationTableIdentifier, deleteWhereClauseSQL) return []string{fallbackUpsertStatement, fallbackDeleteStatement} @@ -560,7 +559,7 @@ func (c *PostgresConnector) generateMergeStatement(destinationTableIdentifier st updateStatements := c.generateUpdateStatement(columnNames, unchangedToastColumns) return fmt.Sprintf(mergeStatementSQL, strings.Join(maps.Values(primaryKeyColumnCasts), ","), - internalSchema, rawTableIdentifier, destinationTableIdentifier, flattenedCastsSQL, + c.metadataSchema, rawTableIdentifier, destinationTableIdentifier, flattenedCastsSQL, strings.Join(primaryKeySelectSQLArray, " AND "), insertColumnsSQL, insertValuesSQL, updateStatements) } diff --git a/flow/connectors/postgres/postgres.go b/flow/connectors/postgres/postgres.go index 6c06cb60c4..fdca5abafa 100644 --- a/flow/connectors/postgres/postgres.go +++ b/flow/connectors/postgres/postgres.go @@ -29,6 +29,7 @@ type PostgresConnector struct { replPool *pgxpool.Pool tableSchemaMapping map[string]*protos.TableSchema customTypesMapping map[uint32]string + metadataSchema string } // NewPostgresConnector creates a new instance of PostgresConnector. @@ -67,11 +68,18 @@ func NewPostgresConnector(ctx context.Context, pgConfig *protos.PostgresConfig) replConnConfig.ConnConfig.RuntimeParams["bytea_output"] = "hex" replConnConfig.MaxConns = 1 + // TODO: replPool not initializing might be intentional, if we only want to use QRep mirrors + // and the user doesn't have the REPLICATION permission replPool, err := pgxpool.NewWithConfig(ctx, replConnConfig) if err != nil { return nil, fmt.Errorf("failed to create connection pool: %w", err) } + metadataSchema := "_peerdb_internal" + if pgConfig.MetadataSchema != nil { + metadataSchema = *pgConfig.MetadataSchema + } + return &PostgresConnector{ connStr: connectionString, ctx: ctx, @@ -79,6 +87,7 @@ func NewPostgresConnector(ctx context.Context, pgConfig *protos.PostgresConfig) pool: pool, replPool: replPool, customTypesMapping: customTypeMap, + metadataSchema: metadataSchema, }, nil } @@ -106,7 +115,7 @@ func (c *PostgresConnector) ConnectionActive() bool { // NeedsSetupMetadataTables returns true if the metadata tables need to be set up. func (c *PostgresConnector) NeedsSetupMetadataTables() bool { result, err := c.tableExists(&utils.SchemaTable{ - Schema: internalSchema, + Schema: c.metadataSchema, Table: mirrorJobsTableIdentifier, }) if err != nil { @@ -128,12 +137,12 @@ func (c *PostgresConnector) SetupMetadataTables() error { } }() - err = c.createInternalSchema(createMetadataTablesTx) + err = c.createMetadataSchema(createMetadataTablesTx) if err != nil { return err } _, err = createMetadataTablesTx.Exec(c.ctx, fmt.Sprintf(createMirrorJobsTableSQL, - internalSchema, mirrorJobsTableIdentifier)) + c.metadataSchema, mirrorJobsTableIdentifier)) if err != nil { return fmt.Errorf("error creating table %s: %w", mirrorJobsTableIdentifier, err) } @@ -148,7 +157,7 @@ func (c *PostgresConnector) SetupMetadataTables() error { // GetLastOffset returns the last synced offset for a job. func (c *PostgresConnector) GetLastOffset(jobName string) (*protos.LastSyncState, error) { rows, err := c.pool. - Query(c.ctx, fmt.Sprintf(getLastOffsetSQL, internalSchema, mirrorJobsTableIdentifier), jobName) + Query(c.ctx, fmt.Sprintf(getLastOffsetSQL, c.metadataSchema, mirrorJobsTableIdentifier), jobName) if err != nil { return nil, fmt.Errorf("error getting last offset for job %s: %w", jobName, err) } @@ -354,7 +363,7 @@ func (c *PostgresConnector) SyncRecords(req *model.SyncRecordsRequest) (*model.S } }() - syncedRecordsCount, err := syncRecordsTx.CopyFrom(c.ctx, pgx.Identifier{internalSchema, rawTableIdentifier}, + syncedRecordsCount, err := syncRecordsTx.CopyFrom(c.ctx, pgx.Identifier{c.metadataSchema, rawTableIdentifier}, []string{"_peerdb_uid", "_peerdb_timestamp", "_peerdb_destination_table_name", "_peerdb_data", "_peerdb_record_type", "_peerdb_match_data", "_peerdb_batch_id", "_peerdb_unchanged_toast_columns"}, pgx.CopyFromRows(records)) @@ -507,21 +516,21 @@ func (c *PostgresConnector) CreateRawTable(req *protos.CreateRawTableInput) (*pr } }() - err = c.createInternalSchema(createRawTableTx) + err = c.createMetadataSchema(createRawTableTx) if err != nil { return nil, fmt.Errorf("error creating internal schema: %w", err) } - _, err = createRawTableTx.Exec(c.ctx, fmt.Sprintf(createRawTableSQL, internalSchema, rawTableIdentifier)) + _, err = createRawTableTx.Exec(c.ctx, fmt.Sprintf(createRawTableSQL, c.metadataSchema, rawTableIdentifier)) if err != nil { return nil, fmt.Errorf("error creating raw table: %w", err) } _, err = createRawTableTx.Exec(c.ctx, fmt.Sprintf(createRawTableBatchIDIndexSQL, rawTableIdentifier, - internalSchema, rawTableIdentifier)) + c.metadataSchema, rawTableIdentifier)) if err != nil { return nil, fmt.Errorf("error creating batch ID index on raw table: %w", err) } _, err = createRawTableTx.Exec(c.ctx, fmt.Sprintf(createRawTableDstTableIndexSQL, rawTableIdentifier, - internalSchema, rawTableIdentifier)) + c.metadataSchema, rawTableIdentifier)) if err != nil { return nil, fmt.Errorf("error creating destion table index on raw table: %w", err) } @@ -723,7 +732,6 @@ func (c *PostgresConnector) ReplayTableSchemaDeltas(flowJobName string, // EnsurePullability ensures that a table is pullable, implementing the Connector interface. func (c *PostgresConnector) EnsurePullability(req *protos.EnsurePullabilityBatchInput, ) (*protos.EnsurePullabilityBatchOutput, error) { - tableIdentifierMapping := make(map[string]*protos.TableIdentifier) for _, tableName := range req.SourceTableIdentifiers { schemaTable, err := utils.ParseSchemaTable(tableName) @@ -838,13 +846,13 @@ func (c *PostgresConnector) SyncFlowCleanup(jobName string) error { } }() - _, err = syncFlowCleanupTx.Exec(c.ctx, fmt.Sprintf(dropTableIfExistsSQL, internalSchema, + _, err = syncFlowCleanupTx.Exec(c.ctx, fmt.Sprintf(dropTableIfExistsSQL, c.metadataSchema, getRawTableIdentifier(jobName))) if err != nil { return fmt.Errorf("unable to drop raw table: %w", err) } _, err = syncFlowCleanupTx.Exec(c.ctx, - fmt.Sprintf(deleteJobMetadataSQL, internalSchema, mirrorJobsTableIdentifier), jobName) + fmt.Sprintf(deleteJobMetadataSQL, c.metadataSchema, mirrorJobsTableIdentifier), jobName) if err != nil { return fmt.Errorf("unable to delete job metadata: %w", err) } diff --git a/flow/connectors/postgres/qrep.go b/flow/connectors/postgres/qrep.go index 2287b81158..26742b4114 100644 --- a/flow/connectors/postgres/qrep.go +++ b/flow/connectors/postgres/qrep.go @@ -553,19 +553,34 @@ func (c *PostgresConnector) SyncQRepRecords( // SetupQRepMetadataTables function for postgres connector func (c *PostgresConnector) SetupQRepMetadataTables(config *protos.QRepConfig) error { - qRepMetadataSchema := `CREATE TABLE IF NOT EXISTS %s ( + createQRepMetadataTableTx, err := c.pool.Begin(c.ctx) + if err != nil { + return fmt.Errorf("error starting transaction for creating qrep metadata table: %w", err) + } + defer func() { + deferErr := createQRepMetadataTableTx.Rollback(c.ctx) + if deferErr != pgx.ErrTxClosed && deferErr != nil { + log.WithFields(log.Fields{ + "flowName": config.FlowJobName, + }).Errorf("unexpected error rolling back transaction for creating qrep metadata table: %v", err) + } + }() + + err = c.createMetadataSchema(createQRepMetadataTableTx) + if err != nil { + return fmt.Errorf("error creating metadata schema: %w", err) + } + + metadataTableIdentifier := pgx.Identifier{c.metadataSchema, qRepMetadataTableName} + createQRepMetadataTableSQL := fmt.Sprintf(`CREATE TABLE IF NOT EXISTS %s( flowJobName TEXT, partitionID TEXT, syncPartition JSONB, syncStartTime TIMESTAMP, syncFinishTime TIMESTAMP DEFAULT NOW() - )` - - // replace table name in schema - qRepMetadataSchema = fmt.Sprintf(qRepMetadataSchema, qRepMetadataTableName) - + )`, metadataTableIdentifier.Sanitize()) // execute create table query - _, err := c.pool.Exec(c.ctx, qRepMetadataSchema) + _, err = createQRepMetadataTableTx.Exec(c.ctx, createQRepMetadataTableSQL) if err != nil { return fmt.Errorf("failed to create table %s: %w", qRepMetadataTableName, err) } @@ -575,12 +590,18 @@ func (c *PostgresConnector) SetupQRepMetadataTables(config *protos.QRepConfig) e if config.WriteMode != nil && config.WriteMode.WriteType == protos.QRepWriteType_QREP_WRITE_MODE_OVERWRITE { - _, err = c.pool.Exec(c.ctx, fmt.Sprintf("TRUNCATE TABLE %s", config.DestinationTableIdentifier)) + _, err = createQRepMetadataTableTx.Exec(c.ctx, + fmt.Sprintf("TRUNCATE TABLE %s", config.DestinationTableIdentifier)) if err != nil { return fmt.Errorf("failed to TRUNCATE table before query replication: %w", err) } } + err = createQRepMetadataTableTx.Commit(c.ctx) + if err != nil { + return fmt.Errorf("error committing transaction for creating qrep metadata table: %w", err) + } + return nil } @@ -612,17 +633,18 @@ func BuildQuery(query string, flowJobName string) (string, error) { // isPartitionSynced checks whether a specific partition is synced func (c *PostgresConnector) isPartitionSynced(partitionID string) (bool, error) { // setup the query string + metadataTableIdentifier := pgx.Identifier{c.metadataSchema, qRepMetadataTableName} queryString := fmt.Sprintf( - "SELECT COUNT(*) FROM %s WHERE partitionID = $1;", - qRepMetadataTableName, + "SELECT COUNT(*)>0 FROM %s WHERE partitionID = $1;", + metadataTableIdentifier.Sanitize(), ) // prepare and execute the query - var count int - err := c.pool.QueryRow(c.ctx, queryString, partitionID).Scan(&count) + var result bool + err := c.pool.QueryRow(c.ctx, queryString, partitionID).Scan(&result) if err != nil { return false, fmt.Errorf("failed to execute query: %w", err) } - return count > 0, nil + return result, nil } diff --git a/flow/connectors/postgres/qrep_sync_method.go b/flow/connectors/postgres/qrep_sync_method.go index f1e818f397..c91d734764 100644 --- a/flow/connectors/postgres/qrep_sync_method.go +++ b/flow/connectors/postgres/qrep_sync_method.go @@ -86,7 +86,7 @@ func (s *QRepStagingTableSync) SyncQRepRecords( } else { // Step 2.1: Create a temp staging table stagingTableName := fmt.Sprintf("_peerdb_staging_%s", util.RandomString(8)) - stagingTableIdentifier := pgx.Identifier{dstTableName.Schema, stagingTableName} + stagingTableIdentifier := pgx.Identifier{s.connector.metadataSchema, stagingTableName} dstTableIdentifier := pgx.Identifier{dstTableName.Schema, dstTableName.Table} createStagingTableStmt := fmt.Sprintf( @@ -109,7 +109,7 @@ func (s *QRepStagingTableSync) SyncQRepRecords( schema.GetColumnNames(), copySource, ) - if err != nil { + if err != nil || numRowsSynced != int64(copySource.NumRecords()) { return -1, fmt.Errorf("failed to copy records into staging table: %v", err) } @@ -173,9 +173,10 @@ func (s *QRepStagingTableSync) SyncQRepRecords( return -1, fmt.Errorf("failed to marshal partition to json: %v", err) } + metadataTableIdentifier := pgx.Identifier{s.connector.metadataSchema, qRepMetadataTableName} insertMetadataStmt := fmt.Sprintf( "INSERT INTO %s VALUES ($1, $2, $3, $4, $5);", - qRepMetadataTableName, + metadataTableIdentifier.Sanitize(), ) log.WithFields(log.Fields{ "flowName": flowJobName, diff --git a/flow/connectors/snowflake/qrep.go b/flow/connectors/snowflake/qrep.go index 094dbcaff9..3fcc8720a0 100644 --- a/flow/connectors/snowflake/qrep.go +++ b/flow/connectors/snowflake/qrep.go @@ -79,7 +79,7 @@ func (c *SnowflakeConnector) createMetadataInsertStatement( `INSERT INTO %s.%s (flowJobName, partitionID, syncPartition, syncStartTime, syncFinishTime) VALUES ('%s', '%s', '%s', '%s'::timestamp, CURRENT_TIMESTAMP);`, - "public", qRepMetadataTableName, jobName, partition.PartitionId, + c.metadataSchema, qRepMetadataTableName, jobName, partition.PartitionId, partitionJSON, startTime.Format(time.RFC3339)) return insertMetadataStmt, nil @@ -111,9 +111,9 @@ func (c *SnowflakeConnector) isPartitionSynced(partitionID string) (bool, error) //nolint:gosec queryString := fmt.Sprintf(` SELECT COUNT(*) - FROM _peerdb_query_replication_metadata + FROM %s.%s WHERE partitionID = '%s' - `, partitionID) + `, c.metadataSchema, qRepMetadataTableName, partitionID) row := c.database.QueryRow(queryString) @@ -126,7 +126,23 @@ func (c *SnowflakeConnector) isPartitionSynced(partitionID string) (bool, error) } func (c *SnowflakeConnector) SetupQRepMetadataTables(config *protos.QRepConfig) error { - err := c.createQRepMetadataTable() + // NOTE that Snowflake does not support transactional DDL + createMetadataTablesTx, err := c.database.BeginTx(c.ctx, nil) + if err != nil { + return fmt.Errorf("unable to begin transaction for creating metadata tables: %w", err) + } + // in case we return after error, ensure transaction is rolled back + defer func() { + deferErr := createMetadataTablesTx.Rollback() + if deferErr != sql.ErrTxDone && deferErr != nil { + log.Errorf("unexpected error while rolling back transaction for creating metadata tables: %v", deferErr) + } + }() + err = c.createPeerDBInternalSchema(createMetadataTablesTx) + if err != nil { + return err + } + err = c.createQRepMetadataTable(createMetadataTablesTx) if err != nil { return err } @@ -145,10 +161,15 @@ func (c *SnowflakeConnector) SetupQRepMetadataTables(config *protos.QRepConfig) } } + err = createMetadataTablesTx.Commit() + if err != nil { + return fmt.Errorf("unable to commit transaction for creating metadata tables: %w", err) + } + return nil } -func (c *SnowflakeConnector) createQRepMetadataTable() error { +func (c *SnowflakeConnector) createQRepMetadataTable(createMetadataTableTx *sql.Tx) error { // Define the schema schemaStatement := ` CREATE TABLE IF NOT EXISTS %s.%s ( @@ -159,12 +180,12 @@ func (c *SnowflakeConnector) createQRepMetadataTable() error { syncFinishTime TIMESTAMP_LTZ ); ` - queryString := fmt.Sprintf(schemaStatement, "public", qRepMetadataTableName) + queryString := fmt.Sprintf(schemaStatement, c.metadataSchema, qRepMetadataTableName) - _, err := c.database.Exec(queryString) + _, err := createMetadataTableTx.Exec(queryString) if err != nil { - log.Errorf("failed to create table %s.%s: %v", "public", qRepMetadataTableName, err) - return fmt.Errorf("failed to create table %s.%s: %w", "public", qRepMetadataTableName, err) + log.Errorf("failed to create table %s.%s: %v", c.metadataSchema, qRepMetadataTableName, err) + return fmt.Errorf("failed to create table %s.%s: %w", c.metadataSchema, qRepMetadataTableName, err) } log.Infof("Created table %s", qRepMetadataTableName) @@ -384,6 +405,5 @@ func (c *SnowflakeConnector) dropStage(stagingPath string, job string) error { } func (c *SnowflakeConnector) getStageNameForJob(job string) string { - // TODO move this from public to peerdb internal schema. - return fmt.Sprintf("public.peerdb_stage_%s", job) + return fmt.Sprintf("%s.peerdb_stage_%s", c.metadataSchema, job) } diff --git a/flow/connectors/snowflake/snowflake.go b/flow/connectors/snowflake/snowflake.go index c31620f82d..4023027949 100644 --- a/flow/connectors/snowflake/snowflake.go +++ b/flow/connectors/snowflake/snowflake.go @@ -24,14 +24,12 @@ import ( //nolint:stylecheck const ( - // all PeerDB specific tables should go in the internal schema. - peerDBInternalSchema = "_PEERDB_INTERNAL" mirrorJobsTableIdentifier = "PEERDB_MIRROR_JOBS" createMirrorJobsTableSQL = `CREATE TABLE IF NOT EXISTS %s.%s(MIRROR_JOB_NAME STRING NOT NULL,OFFSET INT NOT NULL, SYNC_BATCH_ID INT NOT NULL,NORMALIZE_BATCH_ID INT NOT NULL)` - rawTablePrefix = "_PEERDB_RAW" - createPeerDBInternalSchemaSQL = "CREATE TRANSIENT SCHEMA IF NOT EXISTS %s" - createRawTableSQL = `CREATE TABLE IF NOT EXISTS %s.%s(_PEERDB_UID STRING NOT NULL, + rawTablePrefix = "_PEERDB_RAW" + createSchemaSQL = "CREATE TRANSIENT SCHEMA IF NOT EXISTS %s" + createRawTableSQL = `CREATE TABLE IF NOT EXISTS %s.%s(_PEERDB_UID STRING NOT NULL, _PEERDB_TIMESTAMP INT NOT NULL,_PEERDB_DESTINATION_TABLE_NAME STRING NOT NULL,_PEERDB_DATA STRING NOT NULL, _PEERDB_RECORD_TYPE INTEGER NOT NULL, _PEERDB_MATCH_DATA STRING,_PEERDB_BATCH_ID INT, _PEERDB_UNCHANGED_TOAST_COLUMNS STRING)` @@ -90,6 +88,7 @@ type SnowflakeConnector struct { ctx context.Context database *sql.DB tableSchemaMapping map[string]*protos.TableSchema + metadataSchema string } type snowflakeRawRecord struct { @@ -157,10 +156,16 @@ func NewSnowflakeConnector(ctx context.Context, return nil, fmt.Errorf("failed to open connection to Snowflake peer: %w", err) } + metadataSchema := "_PEERDB_INTERNAL" + if snowflakeProtoConfig.MetadataSchema != nil { + metadataSchema = *snowflakeProtoConfig.MetadataSchema + } + return &SnowflakeConnector{ ctx: ctx, database: database, tableSchemaMapping: nil, + metadataSchema: metadataSchema, }, nil } @@ -184,7 +189,7 @@ func (c *SnowflakeConnector) ConnectionActive() bool { } func (c *SnowflakeConnector) NeedsSetupMetadataTables() bool { - result, err := c.checkIfTableExists(peerDBInternalSchema, mirrorJobsTableIdentifier) + result, err := c.checkIfTableExists(c.metadataSchema, mirrorJobsTableIdentifier) if err != nil { return true } @@ -197,12 +202,20 @@ func (c *SnowflakeConnector) SetupMetadataTables() error { if err != nil { return fmt.Errorf("unable to begin transaction for creating metadata tables: %w", err) } + // in case we return after error, ensure transaction is rolled back + defer func() { + deferErr := createMetadataTablesTx.Rollback() + if deferErr != sql.ErrTxDone && deferErr != nil { + log.Errorf("unexpected error while rolling back transaction for creating metadata tables: %v", deferErr) + } + }() + err = c.createPeerDBInternalSchema(createMetadataTablesTx) if err != nil { return err } _, err = createMetadataTablesTx.ExecContext(c.ctx, fmt.Sprintf(createMirrorJobsTableSQL, - peerDBInternalSchema, mirrorJobsTableIdentifier)) + c.metadataSchema, mirrorJobsTableIdentifier)) if err != nil { return fmt.Errorf("error while setting up mirror jobs table: %w", err) } @@ -275,7 +288,7 @@ func (c *SnowflakeConnector) getTableSchemaForTable(tableName string) (*protos.T func (c *SnowflakeConnector) GetLastOffset(jobName string) (*protos.LastSyncState, error) { rows, err := c.database.QueryContext(c.ctx, fmt.Sprintf(getLastOffsetSQL, - peerDBInternalSchema, mirrorJobsTableIdentifier), jobName) + c.metadataSchema, mirrorJobsTableIdentifier), jobName) if err != nil { return nil, fmt.Errorf("error querying Snowflake peer for last syncedID: %w", err) } @@ -306,7 +319,7 @@ func (c *SnowflakeConnector) GetLastOffset(jobName string) (*protos.LastSyncStat } func (c *SnowflakeConnector) GetLastSyncBatchID(jobName string) (int64, error) { - rows, err := c.database.QueryContext(c.ctx, fmt.Sprintf(getLastSyncBatchID_SQL, peerDBInternalSchema, + rows, err := c.database.QueryContext(c.ctx, fmt.Sprintf(getLastSyncBatchID_SQL, c.metadataSchema, mirrorJobsTableIdentifier), jobName) if err != nil { return 0, fmt.Errorf("error querying Snowflake peer for last syncBatchId: %w", err) @@ -325,7 +338,7 @@ func (c *SnowflakeConnector) GetLastSyncBatchID(jobName string) (int64, error) { } func (c *SnowflakeConnector) GetLastNormalizeBatchID(jobName string) (int64, error) { - rows, err := c.database.QueryContext(c.ctx, fmt.Sprintf(getLastNormalizeBatchID_SQL, peerDBInternalSchema, + rows, err := c.database.QueryContext(c.ctx, fmt.Sprintf(getLastNormalizeBatchID_SQL, c.metadataSchema, mirrorJobsTableIdentifier), jobName) if err != nil { return 0, fmt.Errorf("error querying Snowflake peer for last normalizeBatchId: %w", err) @@ -347,7 +360,7 @@ func (c *SnowflakeConnector) getDistinctTableNamesInBatch(flowJobName string, sy normalizeBatchID int64) ([]string, error) { rawTableIdentifier := getRawTableIdentifier(flowJobName) - rows, err := c.database.QueryContext(c.ctx, fmt.Sprintf(getDistinctDestinationTableNames, peerDBInternalSchema, + rows, err := c.database.QueryContext(c.ctx, fmt.Sprintf(getDistinctDestinationTableNames, c.metadataSchema, rawTableIdentifier, normalizeBatchID, syncBatchID)) if err != nil { return nil, fmt.Errorf("error while retrieving table names for normalization: %w", err) @@ -369,7 +382,7 @@ func (c *SnowflakeConnector) getTableNametoUnchangedCols(flowJobName string, syn normalizeBatchID int64) (map[string][]string, error) { rawTableIdentifier := getRawTableIdentifier(flowJobName) - rows, err := c.database.QueryContext(c.ctx, fmt.Sprintf(getTableNametoUnchangedColsSQL, peerDBInternalSchema, + rows, err := c.database.QueryContext(c.ctx, fmt.Sprintf(getTableNametoUnchangedColsSQL, c.metadataSchema, rawTableIdentifier, normalizeBatchID, syncBatchID)) if err != nil { return nil, fmt.Errorf("error while retrieving table names for normalization: %w", err) @@ -658,7 +671,7 @@ func (c *SnowflakeConnector) syncRecordsViaAvro( qrepConfig := &protos.QRepConfig{ StagingPath: "", FlowJobName: req.FlowJobName, - DestinationTableIdentifier: fmt.Sprintf("%s.%s", peerDBInternalSchema, + DestinationTableIdentifier: fmt.Sprintf("%s.%s", c.metadataSchema, rawTableIdentifier), } avroSyncer := NewSnowflakeAvroSyncMethod(qrepConfig, c) @@ -788,7 +801,7 @@ func (c *SnowflakeConnector) CreateRawTable(req *protos.CreateRawTableInput) (*p // there is no easy way to check if a table has the same schema in Snowflake, // so just executing the CREATE TABLE IF NOT EXISTS blindly. _, err = createRawTableTx.ExecContext(c.ctx, - fmt.Sprintf(createRawTableSQL, peerDBInternalSchema, rawTableIdentifier)) + fmt.Sprintf(createRawTableSQL, c.metadataSchema, rawTableIdentifier)) if err != nil { return nil, fmt.Errorf("unable to create raw table: %w", err) } @@ -824,7 +837,7 @@ func (c *SnowflakeConnector) SyncFlowCleanup(jobName string) error { } }() - row := syncFlowCleanupTx.QueryRowContext(c.ctx, checkSchemaExistsSQL, peerDBInternalSchema) + row := syncFlowCleanupTx.QueryRowContext(c.ctx, checkSchemaExistsSQL, c.metadataSchema) var schemaExists bool err = row.Scan(&schemaExists) if err != nil { @@ -832,13 +845,13 @@ func (c *SnowflakeConnector) SyncFlowCleanup(jobName string) error { } if schemaExists { - _, err = syncFlowCleanupTx.ExecContext(c.ctx, fmt.Sprintf(dropTableIfExistsSQL, peerDBInternalSchema, + _, err = syncFlowCleanupTx.ExecContext(c.ctx, fmt.Sprintf(dropTableIfExistsSQL, c.metadataSchema, getRawTableIdentifier(jobName))) if err != nil { return fmt.Errorf("unable to drop raw table: %w", err) } _, err = syncFlowCleanupTx.ExecContext(c.ctx, - fmt.Sprintf(deleteJobMetadataSQL, peerDBInternalSchema, mirrorJobsTableIdentifier), jobName) + fmt.Sprintf(deleteJobMetadataSQL, c.metadataSchema, mirrorJobsTableIdentifier), jobName) if err != nil { return fmt.Errorf("unable to delete job metadata: %w", err) } @@ -906,11 +919,11 @@ func generateCreateTableSQLForNormalizedTable( strings.TrimSuffix(strings.Join(createTableSQLArray, ""), ",")) } -func generateMultiValueInsertSQL(tableIdentifier string, chunkSize int) string { +func generateMultiValueInsertSQL(metadataSchema string, tableIdentifier string, chunkSize int) string { // inferring the width of the raw table from the create table statement rawTableWidth := strings.Count(createRawTableSQL, ",") + 1 - return fmt.Sprintf(rawTableMultiValueInsertSQL, peerDBInternalSchema, tableIdentifier, + return fmt.Sprintf(rawTableMultiValueInsertSQL, metadataSchema, tableIdentifier, strings.TrimSuffix(strings.Repeat(fmt.Sprintf("(%s),", strings.TrimSuffix(strings.Repeat("?,", rawTableWidth), ",")), chunkSize), ",")) } @@ -929,7 +942,7 @@ func (c *SnowflakeConnector) insertRecordsInRawTable(rawTableIdentifier string, record.data, record.recordType, record.matchData, record.batchID, record.unchangedToastColumns) } _, err := syncRecordsTx.ExecContext(c.ctx, - generateMultiValueInsertSQL(rawTableIdentifier, len(snowflakeRawRecords)), rawRecordsData...) + generateMultiValueInsertSQL(c.metadataSchema, rawTableIdentifier, len(snowflakeRawRecords)), rawRecordsData...) if err != nil { return fmt.Errorf("failed to insert record into raw table: %w", err) } @@ -1039,7 +1052,7 @@ func parseTableName(tableName string) (*tableNameComponents, error) { func (c *SnowflakeConnector) jobMetadataExists(jobName string) (bool, error) { rows, err := c.database.QueryContext(c.ctx, - fmt.Sprintf(checkIfJobMetadataExistsSQL, peerDBInternalSchema, mirrorJobsTableIdentifier), jobName) + fmt.Sprintf(checkIfJobMetadataExistsSQL, c.metadataSchema, mirrorJobsTableIdentifier), jobName) if err != nil { return false, fmt.Errorf("failed to check if job exists: %w", err) } @@ -1062,14 +1075,14 @@ func (c *SnowflakeConnector) updateSyncMetadata(flowJobName string, lastCP int64 if !jobMetadataExists { _, err := syncRecordsTx.ExecContext(c.ctx, - fmt.Sprintf(insertJobMetadataSQL, peerDBInternalSchema, mirrorJobsTableIdentifier), + fmt.Sprintf(insertJobMetadataSQL, c.metadataSchema, mirrorJobsTableIdentifier), flowJobName, lastCP, syncBatchID, 0) if err != nil { return fmt.Errorf("failed to insert flow job status: %w", err) } } else { _, err := syncRecordsTx.ExecContext(c.ctx, - fmt.Sprintf(updateMetadataForSyncRecordsSQL, peerDBInternalSchema, mirrorJobsTableIdentifier), + fmt.Sprintf(updateMetadataForSyncRecordsSQL, c.metadataSchema, mirrorJobsTableIdentifier), lastCP, syncBatchID, flowJobName) if err != nil { return fmt.Errorf("failed to update flow job status: %w", err) @@ -1090,7 +1103,7 @@ func (c *SnowflakeConnector) updateNormalizeMetadata(flowJobName string, } _, err = normalizeRecordsTx.ExecContext(c.ctx, - fmt.Sprintf(updateMetadataForNormalizeRecordsSQL, peerDBInternalSchema, mirrorJobsTableIdentifier), + fmt.Sprintf(updateMetadataForNormalizeRecordsSQL, c.metadataSchema, mirrorJobsTableIdentifier), normalizeBatchID, flowJobName) if err != nil { return fmt.Errorf("failed to update metadata for NormalizeTables: %w", err) @@ -1100,7 +1113,7 @@ func (c *SnowflakeConnector) updateNormalizeMetadata(flowJobName string, } func (c *SnowflakeConnector) createPeerDBInternalSchema(createSchemaTx *sql.Tx) error { - _, err := createSchemaTx.ExecContext(c.ctx, fmt.Sprintf(createPeerDBInternalSchemaSQL, peerDBInternalSchema)) + _, err := createSchemaTx.ExecContext(c.ctx, fmt.Sprintf(createSchemaSQL, c.metadataSchema)) if err != nil { return fmt.Errorf("error while creating internal schema for PeerDB: %w", err) } diff --git a/flow/generated/protos/peers.pb.go b/flow/generated/protos/peers.pb.go index 7f8bb4a371..6ac88a4c11 100644 --- a/flow/generated/protos/peers.pb.go +++ b/flow/generated/protos/peers.pb.go @@ -98,6 +98,8 @@ type SnowflakeConfig struct { QueryTimeout uint64 `protobuf:"varint,8,opt,name=query_timeout,json=queryTimeout,proto3" json:"query_timeout,omitempty"` S3Integration string `protobuf:"bytes,9,opt,name=s3_integration,json=s3Integration,proto3" json:"s3_integration,omitempty"` Password *string `protobuf:"bytes,10,opt,name=password,proto3,oneof" json:"password,omitempty"` + // defaults to _PEERDB_INTERNAL + MetadataSchema *string `protobuf:"bytes,11,opt,name=metadata_schema,json=metadataSchema,proto3,oneof" json:"metadata_schema,omitempty"` } func (x *SnowflakeConfig) Reset() { @@ -195,6 +197,13 @@ func (x *SnowflakeConfig) GetPassword() string { return "" } +func (x *SnowflakeConfig) GetMetadataSchema() string { + if x != nil && x.MetadataSchema != nil { + return *x.MetadataSchema + } + return "" +} + type BigqueryConfig struct { state protoimpl.MessageState sizeCache protoimpl.SizeCache @@ -413,6 +422,8 @@ type PostgresConfig struct { Database string `protobuf:"bytes,5,opt,name=database,proto3" json:"database,omitempty"` // this is used only in query replication mode right now. TransactionSnapshot string `protobuf:"bytes,6,opt,name=transaction_snapshot,json=transactionSnapshot,proto3" json:"transaction_snapshot,omitempty"` + // defaults to _peerdb_internal + MetadataSchema *string `protobuf:"bytes,7,opt,name=metadata_schema,json=metadataSchema,proto3,oneof" json:"metadata_schema,omitempty"` } func (x *PostgresConfig) Reset() { @@ -489,6 +500,13 @@ func (x *PostgresConfig) GetTransactionSnapshot() string { return "" } +func (x *PostgresConfig) GetMetadataSchema() string { + if x != nil && x.MetadataSchema != nil { + return *x.MetadataSchema + } + return "" +} + type EventHubConfig struct { state protoimpl.MessageState sizeCache protoimpl.SizeCache @@ -1010,7 +1028,7 @@ var File_peers_proto protoreflect.FileDescriptor var file_peers_proto_rawDesc = []byte{ 0x0a, 0x0b, 0x70, 0x65, 0x65, 0x72, 0x73, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x12, 0x0c, 0x70, - 0x65, 0x65, 0x72, 0x64, 0x62, 0x5f, 0x70, 0x65, 0x65, 0x72, 0x73, 0x22, 0xb5, 0x02, 0x0a, 0x0f, + 0x65, 0x65, 0x72, 0x64, 0x62, 0x5f, 0x70, 0x65, 0x65, 0x72, 0x73, 0x22, 0xf7, 0x02, 0x0a, 0x0f, 0x53, 0x6e, 0x6f, 0x77, 0x66, 0x6c, 0x61, 0x6b, 0x65, 0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x12, 0x1d, 0x0a, 0x0a, 0x61, 0x63, 0x63, 0x6f, 0x75, 0x6e, 0x74, 0x5f, 0x69, 0x64, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x09, 0x61, 0x63, 0x63, 0x6f, 0x75, 0x6e, 0x74, 0x49, 0x64, 0x12, 0x1a, @@ -1029,182 +1047,190 @@ var file_peers_proto_rawDesc = []byte{ 0x18, 0x09, 0x20, 0x01, 0x28, 0x09, 0x52, 0x0d, 0x73, 0x33, 0x49, 0x6e, 0x74, 0x65, 0x67, 0x72, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x12, 0x1f, 0x0a, 0x08, 0x70, 0x61, 0x73, 0x73, 0x77, 0x6f, 0x72, 0x64, 0x18, 0x0a, 0x20, 0x01, 0x28, 0x09, 0x48, 0x00, 0x52, 0x08, 0x70, 0x61, 0x73, 0x73, 0x77, - 0x6f, 0x72, 0x64, 0x88, 0x01, 0x01, 0x42, 0x0b, 0x0a, 0x09, 0x5f, 0x70, 0x61, 0x73, 0x73, 0x77, - 0x6f, 0x72, 0x64, 0x22, 0x99, 0x03, 0x0a, 0x0e, 0x42, 0x69, 0x67, 0x71, 0x75, 0x65, 0x72, 0x79, - 0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x12, 0x1b, 0x0a, 0x09, 0x61, 0x75, 0x74, 0x68, 0x5f, 0x74, - 0x79, 0x70, 0x65, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x08, 0x61, 0x75, 0x74, 0x68, 0x54, - 0x79, 0x70, 0x65, 0x12, 0x1d, 0x0a, 0x0a, 0x70, 0x72, 0x6f, 0x6a, 0x65, 0x63, 0x74, 0x5f, 0x69, - 0x64, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x09, 0x70, 0x72, 0x6f, 0x6a, 0x65, 0x63, 0x74, - 0x49, 0x64, 0x12, 0x24, 0x0a, 0x0e, 0x70, 0x72, 0x69, 0x76, 0x61, 0x74, 0x65, 0x5f, 0x6b, 0x65, - 0x79, 0x5f, 0x69, 0x64, 0x18, 0x03, 0x20, 0x01, 0x28, 0x09, 0x52, 0x0c, 0x70, 0x72, 0x69, 0x76, - 0x61, 0x74, 0x65, 0x4b, 0x65, 0x79, 0x49, 0x64, 0x12, 0x1f, 0x0a, 0x0b, 0x70, 0x72, 0x69, 0x76, - 0x61, 0x74, 0x65, 0x5f, 0x6b, 0x65, 0x79, 0x18, 0x04, 0x20, 0x01, 0x28, 0x09, 0x52, 0x0a, 0x70, - 0x72, 0x69, 0x76, 0x61, 0x74, 0x65, 0x4b, 0x65, 0x79, 0x12, 0x21, 0x0a, 0x0c, 0x63, 0x6c, 0x69, - 0x65, 0x6e, 0x74, 0x5f, 0x65, 0x6d, 0x61, 0x69, 0x6c, 0x18, 0x05, 0x20, 0x01, 0x28, 0x09, 0x52, - 0x0b, 0x63, 0x6c, 0x69, 0x65, 0x6e, 0x74, 0x45, 0x6d, 0x61, 0x69, 0x6c, 0x12, 0x1b, 0x0a, 0x09, - 0x63, 0x6c, 0x69, 0x65, 0x6e, 0x74, 0x5f, 0x69, 0x64, 0x18, 0x06, 0x20, 0x01, 0x28, 0x09, 0x52, - 0x08, 0x63, 0x6c, 0x69, 0x65, 0x6e, 0x74, 0x49, 0x64, 0x12, 0x19, 0x0a, 0x08, 0x61, 0x75, 0x74, - 0x68, 0x5f, 0x75, 0x72, 0x69, 0x18, 0x07, 0x20, 0x01, 0x28, 0x09, 0x52, 0x07, 0x61, 0x75, 0x74, - 0x68, 0x55, 0x72, 0x69, 0x12, 0x1b, 0x0a, 0x09, 0x74, 0x6f, 0x6b, 0x65, 0x6e, 0x5f, 0x75, 0x72, - 0x69, 0x18, 0x08, 0x20, 0x01, 0x28, 0x09, 0x52, 0x08, 0x74, 0x6f, 0x6b, 0x65, 0x6e, 0x55, 0x72, - 0x69, 0x12, 0x3c, 0x0a, 0x1b, 0x61, 0x75, 0x74, 0x68, 0x5f, 0x70, 0x72, 0x6f, 0x76, 0x69, 0x64, - 0x65, 0x72, 0x5f, 0x78, 0x35, 0x30, 0x39, 0x5f, 0x63, 0x65, 0x72, 0x74, 0x5f, 0x75, 0x72, 0x6c, - 0x18, 0x09, 0x20, 0x01, 0x28, 0x09, 0x52, 0x17, 0x61, 0x75, 0x74, 0x68, 0x50, 0x72, 0x6f, 0x76, - 0x69, 0x64, 0x65, 0x72, 0x58, 0x35, 0x30, 0x39, 0x43, 0x65, 0x72, 0x74, 0x55, 0x72, 0x6c, 0x12, - 0x2f, 0x0a, 0x14, 0x63, 0x6c, 0x69, 0x65, 0x6e, 0x74, 0x5f, 0x78, 0x35, 0x30, 0x39, 0x5f, 0x63, - 0x65, 0x72, 0x74, 0x5f, 0x75, 0x72, 0x6c, 0x18, 0x0a, 0x20, 0x01, 0x28, 0x09, 0x52, 0x11, 0x63, - 0x6c, 0x69, 0x65, 0x6e, 0x74, 0x58, 0x35, 0x30, 0x39, 0x43, 0x65, 0x72, 0x74, 0x55, 0x72, 0x6c, - 0x12, 0x1d, 0x0a, 0x0a, 0x64, 0x61, 0x74, 0x61, 0x73, 0x65, 0x74, 0x5f, 0x69, 0x64, 0x18, 0x0b, - 0x20, 0x01, 0x28, 0x09, 0x52, 0x09, 0x64, 0x61, 0x74, 0x61, 0x73, 0x65, 0x74, 0x49, 0x64, 0x22, - 0xa3, 0x01, 0x0a, 0x0b, 0x4d, 0x6f, 0x6e, 0x67, 0x6f, 0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x12, - 0x1a, 0x0a, 0x08, 0x75, 0x73, 0x65, 0x72, 0x6e, 0x61, 0x6d, 0x65, 0x18, 0x01, 0x20, 0x01, 0x28, - 0x09, 0x52, 0x08, 0x75, 0x73, 0x65, 0x72, 0x6e, 0x61, 0x6d, 0x65, 0x12, 0x1a, 0x0a, 0x08, 0x70, - 0x61, 0x73, 0x73, 0x77, 0x6f, 0x72, 0x64, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x08, 0x70, - 0x61, 0x73, 0x73, 0x77, 0x6f, 0x72, 0x64, 0x12, 0x1e, 0x0a, 0x0a, 0x63, 0x6c, 0x75, 0x73, 0x74, - 0x65, 0x72, 0x75, 0x72, 0x6c, 0x18, 0x03, 0x20, 0x01, 0x28, 0x09, 0x52, 0x0a, 0x63, 0x6c, 0x75, - 0x73, 0x74, 0x65, 0x72, 0x75, 0x72, 0x6c, 0x12, 0x20, 0x0a, 0x0b, 0x63, 0x6c, 0x75, 0x73, 0x74, - 0x65, 0x72, 0x70, 0x6f, 0x72, 0x74, 0x18, 0x04, 0x20, 0x01, 0x28, 0x05, 0x52, 0x0b, 0x63, 0x6c, - 0x75, 0x73, 0x74, 0x65, 0x72, 0x70, 0x6f, 0x72, 0x74, 0x12, 0x1a, 0x0a, 0x08, 0x64, 0x61, 0x74, - 0x61, 0x62, 0x61, 0x73, 0x65, 0x18, 0x05, 0x20, 0x01, 0x28, 0x09, 0x52, 0x08, 0x64, 0x61, 0x74, - 0x61, 0x62, 0x61, 0x73, 0x65, 0x22, 0xb7, 0x01, 0x0a, 0x0e, 0x50, 0x6f, 0x73, 0x74, 0x67, 0x72, - 0x65, 0x73, 0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x12, 0x12, 0x0a, 0x04, 0x68, 0x6f, 0x73, 0x74, - 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x04, 0x68, 0x6f, 0x73, 0x74, 0x12, 0x12, 0x0a, 0x04, + 0x6f, 0x72, 0x64, 0x88, 0x01, 0x01, 0x12, 0x2c, 0x0a, 0x0f, 0x6d, 0x65, 0x74, 0x61, 0x64, 0x61, + 0x74, 0x61, 0x5f, 0x73, 0x63, 0x68, 0x65, 0x6d, 0x61, 0x18, 0x0b, 0x20, 0x01, 0x28, 0x09, 0x48, + 0x01, 0x52, 0x0e, 0x6d, 0x65, 0x74, 0x61, 0x64, 0x61, 0x74, 0x61, 0x53, 0x63, 0x68, 0x65, 0x6d, + 0x61, 0x88, 0x01, 0x01, 0x42, 0x0b, 0x0a, 0x09, 0x5f, 0x70, 0x61, 0x73, 0x73, 0x77, 0x6f, 0x72, + 0x64, 0x42, 0x12, 0x0a, 0x10, 0x5f, 0x6d, 0x65, 0x74, 0x61, 0x64, 0x61, 0x74, 0x61, 0x5f, 0x73, + 0x63, 0x68, 0x65, 0x6d, 0x61, 0x22, 0x99, 0x03, 0x0a, 0x0e, 0x42, 0x69, 0x67, 0x71, 0x75, 0x65, + 0x72, 0x79, 0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x12, 0x1b, 0x0a, 0x09, 0x61, 0x75, 0x74, 0x68, + 0x5f, 0x74, 0x79, 0x70, 0x65, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x08, 0x61, 0x75, 0x74, + 0x68, 0x54, 0x79, 0x70, 0x65, 0x12, 0x1d, 0x0a, 0x0a, 0x70, 0x72, 0x6f, 0x6a, 0x65, 0x63, 0x74, + 0x5f, 0x69, 0x64, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x09, 0x70, 0x72, 0x6f, 0x6a, 0x65, + 0x63, 0x74, 0x49, 0x64, 0x12, 0x24, 0x0a, 0x0e, 0x70, 0x72, 0x69, 0x76, 0x61, 0x74, 0x65, 0x5f, + 0x6b, 0x65, 0x79, 0x5f, 0x69, 0x64, 0x18, 0x03, 0x20, 0x01, 0x28, 0x09, 0x52, 0x0c, 0x70, 0x72, + 0x69, 0x76, 0x61, 0x74, 0x65, 0x4b, 0x65, 0x79, 0x49, 0x64, 0x12, 0x1f, 0x0a, 0x0b, 0x70, 0x72, + 0x69, 0x76, 0x61, 0x74, 0x65, 0x5f, 0x6b, 0x65, 0x79, 0x18, 0x04, 0x20, 0x01, 0x28, 0x09, 0x52, + 0x0a, 0x70, 0x72, 0x69, 0x76, 0x61, 0x74, 0x65, 0x4b, 0x65, 0x79, 0x12, 0x21, 0x0a, 0x0c, 0x63, + 0x6c, 0x69, 0x65, 0x6e, 0x74, 0x5f, 0x65, 0x6d, 0x61, 0x69, 0x6c, 0x18, 0x05, 0x20, 0x01, 0x28, + 0x09, 0x52, 0x0b, 0x63, 0x6c, 0x69, 0x65, 0x6e, 0x74, 0x45, 0x6d, 0x61, 0x69, 0x6c, 0x12, 0x1b, + 0x0a, 0x09, 0x63, 0x6c, 0x69, 0x65, 0x6e, 0x74, 0x5f, 0x69, 0x64, 0x18, 0x06, 0x20, 0x01, 0x28, + 0x09, 0x52, 0x08, 0x63, 0x6c, 0x69, 0x65, 0x6e, 0x74, 0x49, 0x64, 0x12, 0x19, 0x0a, 0x08, 0x61, + 0x75, 0x74, 0x68, 0x5f, 0x75, 0x72, 0x69, 0x18, 0x07, 0x20, 0x01, 0x28, 0x09, 0x52, 0x07, 0x61, + 0x75, 0x74, 0x68, 0x55, 0x72, 0x69, 0x12, 0x1b, 0x0a, 0x09, 0x74, 0x6f, 0x6b, 0x65, 0x6e, 0x5f, + 0x75, 0x72, 0x69, 0x18, 0x08, 0x20, 0x01, 0x28, 0x09, 0x52, 0x08, 0x74, 0x6f, 0x6b, 0x65, 0x6e, + 0x55, 0x72, 0x69, 0x12, 0x3c, 0x0a, 0x1b, 0x61, 0x75, 0x74, 0x68, 0x5f, 0x70, 0x72, 0x6f, 0x76, + 0x69, 0x64, 0x65, 0x72, 0x5f, 0x78, 0x35, 0x30, 0x39, 0x5f, 0x63, 0x65, 0x72, 0x74, 0x5f, 0x75, + 0x72, 0x6c, 0x18, 0x09, 0x20, 0x01, 0x28, 0x09, 0x52, 0x17, 0x61, 0x75, 0x74, 0x68, 0x50, 0x72, + 0x6f, 0x76, 0x69, 0x64, 0x65, 0x72, 0x58, 0x35, 0x30, 0x39, 0x43, 0x65, 0x72, 0x74, 0x55, 0x72, + 0x6c, 0x12, 0x2f, 0x0a, 0x14, 0x63, 0x6c, 0x69, 0x65, 0x6e, 0x74, 0x5f, 0x78, 0x35, 0x30, 0x39, + 0x5f, 0x63, 0x65, 0x72, 0x74, 0x5f, 0x75, 0x72, 0x6c, 0x18, 0x0a, 0x20, 0x01, 0x28, 0x09, 0x52, + 0x11, 0x63, 0x6c, 0x69, 0x65, 0x6e, 0x74, 0x58, 0x35, 0x30, 0x39, 0x43, 0x65, 0x72, 0x74, 0x55, + 0x72, 0x6c, 0x12, 0x1d, 0x0a, 0x0a, 0x64, 0x61, 0x74, 0x61, 0x73, 0x65, 0x74, 0x5f, 0x69, 0x64, + 0x18, 0x0b, 0x20, 0x01, 0x28, 0x09, 0x52, 0x09, 0x64, 0x61, 0x74, 0x61, 0x73, 0x65, 0x74, 0x49, + 0x64, 0x22, 0xa3, 0x01, 0x0a, 0x0b, 0x4d, 0x6f, 0x6e, 0x67, 0x6f, 0x43, 0x6f, 0x6e, 0x66, 0x69, + 0x67, 0x12, 0x1a, 0x0a, 0x08, 0x75, 0x73, 0x65, 0x72, 0x6e, 0x61, 0x6d, 0x65, 0x18, 0x01, 0x20, + 0x01, 0x28, 0x09, 0x52, 0x08, 0x75, 0x73, 0x65, 0x72, 0x6e, 0x61, 0x6d, 0x65, 0x12, 0x1a, 0x0a, + 0x08, 0x70, 0x61, 0x73, 0x73, 0x77, 0x6f, 0x72, 0x64, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, + 0x08, 0x70, 0x61, 0x73, 0x73, 0x77, 0x6f, 0x72, 0x64, 0x12, 0x1e, 0x0a, 0x0a, 0x63, 0x6c, 0x75, + 0x73, 0x74, 0x65, 0x72, 0x75, 0x72, 0x6c, 0x18, 0x03, 0x20, 0x01, 0x28, 0x09, 0x52, 0x0a, 0x63, + 0x6c, 0x75, 0x73, 0x74, 0x65, 0x72, 0x75, 0x72, 0x6c, 0x12, 0x20, 0x0a, 0x0b, 0x63, 0x6c, 0x75, + 0x73, 0x74, 0x65, 0x72, 0x70, 0x6f, 0x72, 0x74, 0x18, 0x04, 0x20, 0x01, 0x28, 0x05, 0x52, 0x0b, + 0x63, 0x6c, 0x75, 0x73, 0x74, 0x65, 0x72, 0x70, 0x6f, 0x72, 0x74, 0x12, 0x1a, 0x0a, 0x08, 0x64, + 0x61, 0x74, 0x61, 0x62, 0x61, 0x73, 0x65, 0x18, 0x05, 0x20, 0x01, 0x28, 0x09, 0x52, 0x08, 0x64, + 0x61, 0x74, 0x61, 0x62, 0x61, 0x73, 0x65, 0x22, 0xf9, 0x01, 0x0a, 0x0e, 0x50, 0x6f, 0x73, 0x74, + 0x67, 0x72, 0x65, 0x73, 0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x12, 0x12, 0x0a, 0x04, 0x68, 0x6f, + 0x73, 0x74, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x04, 0x68, 0x6f, 0x73, 0x74, 0x12, 0x12, + 0x0a, 0x04, 0x70, 0x6f, 0x72, 0x74, 0x18, 0x02, 0x20, 0x01, 0x28, 0x0d, 0x52, 0x04, 0x70, 0x6f, + 0x72, 0x74, 0x12, 0x12, 0x0a, 0x04, 0x75, 0x73, 0x65, 0x72, 0x18, 0x03, 0x20, 0x01, 0x28, 0x09, + 0x52, 0x04, 0x75, 0x73, 0x65, 0x72, 0x12, 0x1a, 0x0a, 0x08, 0x70, 0x61, 0x73, 0x73, 0x77, 0x6f, + 0x72, 0x64, 0x18, 0x04, 0x20, 0x01, 0x28, 0x09, 0x52, 0x08, 0x70, 0x61, 0x73, 0x73, 0x77, 0x6f, + 0x72, 0x64, 0x12, 0x1a, 0x0a, 0x08, 0x64, 0x61, 0x74, 0x61, 0x62, 0x61, 0x73, 0x65, 0x18, 0x05, + 0x20, 0x01, 0x28, 0x09, 0x52, 0x08, 0x64, 0x61, 0x74, 0x61, 0x62, 0x61, 0x73, 0x65, 0x12, 0x31, + 0x0a, 0x14, 0x74, 0x72, 0x61, 0x6e, 0x73, 0x61, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x5f, 0x73, 0x6e, + 0x61, 0x70, 0x73, 0x68, 0x6f, 0x74, 0x18, 0x06, 0x20, 0x01, 0x28, 0x09, 0x52, 0x13, 0x74, 0x72, + 0x61, 0x6e, 0x73, 0x61, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x53, 0x6e, 0x61, 0x70, 0x73, 0x68, 0x6f, + 0x74, 0x12, 0x2c, 0x0a, 0x0f, 0x6d, 0x65, 0x74, 0x61, 0x64, 0x61, 0x74, 0x61, 0x5f, 0x73, 0x63, + 0x68, 0x65, 0x6d, 0x61, 0x18, 0x07, 0x20, 0x01, 0x28, 0x09, 0x48, 0x00, 0x52, 0x0e, 0x6d, 0x65, + 0x74, 0x61, 0x64, 0x61, 0x74, 0x61, 0x53, 0x63, 0x68, 0x65, 0x6d, 0x61, 0x88, 0x01, 0x01, 0x42, + 0x12, 0x0a, 0x10, 0x5f, 0x6d, 0x65, 0x74, 0x61, 0x64, 0x61, 0x74, 0x61, 0x5f, 0x73, 0x63, 0x68, + 0x65, 0x6d, 0x61, 0x22, 0xbd, 0x02, 0x0a, 0x0e, 0x45, 0x76, 0x65, 0x6e, 0x74, 0x48, 0x75, 0x62, + 0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x12, 0x1c, 0x0a, 0x09, 0x6e, 0x61, 0x6d, 0x65, 0x73, 0x70, + 0x61, 0x63, 0x65, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x09, 0x6e, 0x61, 0x6d, 0x65, 0x73, + 0x70, 0x61, 0x63, 0x65, 0x12, 0x25, 0x0a, 0x0e, 0x72, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, + 0x5f, 0x67, 0x72, 0x6f, 0x75, 0x70, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x0d, 0x72, 0x65, + 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x47, 0x72, 0x6f, 0x75, 0x70, 0x12, 0x1a, 0x0a, 0x08, 0x6c, + 0x6f, 0x63, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x18, 0x03, 0x20, 0x01, 0x28, 0x09, 0x52, 0x08, 0x6c, + 0x6f, 0x63, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x12, 0x3d, 0x0a, 0x0b, 0x6d, 0x65, 0x74, 0x61, 0x64, + 0x61, 0x74, 0x61, 0x5f, 0x64, 0x62, 0x18, 0x04, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x1c, 0x2e, 0x70, + 0x65, 0x65, 0x72, 0x64, 0x62, 0x5f, 0x70, 0x65, 0x65, 0x72, 0x73, 0x2e, 0x50, 0x6f, 0x73, 0x74, + 0x67, 0x72, 0x65, 0x73, 0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x52, 0x0a, 0x6d, 0x65, 0x74, 0x61, + 0x64, 0x61, 0x74, 0x61, 0x44, 0x62, 0x12, 0x27, 0x0a, 0x0f, 0x73, 0x75, 0x62, 0x73, 0x63, 0x72, + 0x69, 0x70, 0x74, 0x69, 0x6f, 0x6e, 0x5f, 0x69, 0x64, 0x18, 0x05, 0x20, 0x01, 0x28, 0x09, 0x52, + 0x0e, 0x73, 0x75, 0x62, 0x73, 0x63, 0x72, 0x69, 0x70, 0x74, 0x69, 0x6f, 0x6e, 0x49, 0x64, 0x12, + 0x27, 0x0a, 0x0f, 0x70, 0x61, 0x72, 0x74, 0x69, 0x74, 0x69, 0x6f, 0x6e, 0x5f, 0x63, 0x6f, 0x75, + 0x6e, 0x74, 0x18, 0x06, 0x20, 0x01, 0x28, 0x0d, 0x52, 0x0e, 0x70, 0x61, 0x72, 0x74, 0x69, 0x74, + 0x69, 0x6f, 0x6e, 0x43, 0x6f, 0x75, 0x6e, 0x74, 0x12, 0x39, 0x0a, 0x19, 0x6d, 0x65, 0x73, 0x73, + 0x61, 0x67, 0x65, 0x5f, 0x72, 0x65, 0x74, 0x65, 0x6e, 0x74, 0x69, 0x6f, 0x6e, 0x5f, 0x69, 0x6e, + 0x5f, 0x64, 0x61, 0x79, 0x73, 0x18, 0x07, 0x20, 0x01, 0x28, 0x0d, 0x52, 0x16, 0x6d, 0x65, 0x73, + 0x73, 0x61, 0x67, 0x65, 0x52, 0x65, 0x74, 0x65, 0x6e, 0x74, 0x69, 0x6f, 0x6e, 0x49, 0x6e, 0x44, + 0x61, 0x79, 0x73, 0x22, 0xa7, 0x02, 0x0a, 0x13, 0x45, 0x76, 0x65, 0x6e, 0x74, 0x48, 0x75, 0x62, + 0x47, 0x72, 0x6f, 0x75, 0x70, 0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x12, 0x4e, 0x0a, 0x09, 0x65, + 0x76, 0x65, 0x6e, 0x74, 0x68, 0x75, 0x62, 0x73, 0x18, 0x01, 0x20, 0x03, 0x28, 0x0b, 0x32, 0x30, + 0x2e, 0x70, 0x65, 0x65, 0x72, 0x64, 0x62, 0x5f, 0x70, 0x65, 0x65, 0x72, 0x73, 0x2e, 0x45, 0x76, + 0x65, 0x6e, 0x74, 0x48, 0x75, 0x62, 0x47, 0x72, 0x6f, 0x75, 0x70, 0x43, 0x6f, 0x6e, 0x66, 0x69, + 0x67, 0x2e, 0x45, 0x76, 0x65, 0x6e, 0x74, 0x68, 0x75, 0x62, 0x73, 0x45, 0x6e, 0x74, 0x72, 0x79, + 0x52, 0x09, 0x65, 0x76, 0x65, 0x6e, 0x74, 0x68, 0x75, 0x62, 0x73, 0x12, 0x3d, 0x0a, 0x0b, 0x6d, + 0x65, 0x74, 0x61, 0x64, 0x61, 0x74, 0x61, 0x5f, 0x64, 0x62, 0x18, 0x02, 0x20, 0x01, 0x28, 0x0b, + 0x32, 0x1c, 0x2e, 0x70, 0x65, 0x65, 0x72, 0x64, 0x62, 0x5f, 0x70, 0x65, 0x65, 0x72, 0x73, 0x2e, + 0x50, 0x6f, 0x73, 0x74, 0x67, 0x72, 0x65, 0x73, 0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x52, 0x0a, + 0x6d, 0x65, 0x74, 0x61, 0x64, 0x61, 0x74, 0x61, 0x44, 0x62, 0x12, 0x25, 0x0a, 0x0e, 0x75, 0x6e, + 0x6e, 0x65, 0x73, 0x74, 0x5f, 0x63, 0x6f, 0x6c, 0x75, 0x6d, 0x6e, 0x73, 0x18, 0x03, 0x20, 0x03, + 0x28, 0x09, 0x52, 0x0d, 0x75, 0x6e, 0x6e, 0x65, 0x73, 0x74, 0x43, 0x6f, 0x6c, 0x75, 0x6d, 0x6e, + 0x73, 0x1a, 0x5a, 0x0a, 0x0e, 0x45, 0x76, 0x65, 0x6e, 0x74, 0x68, 0x75, 0x62, 0x73, 0x45, 0x6e, + 0x74, 0x72, 0x79, 0x12, 0x10, 0x0a, 0x03, 0x6b, 0x65, 0x79, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, + 0x52, 0x03, 0x6b, 0x65, 0x79, 0x12, 0x32, 0x0a, 0x05, 0x76, 0x61, 0x6c, 0x75, 0x65, 0x18, 0x02, + 0x20, 0x01, 0x28, 0x0b, 0x32, 0x1c, 0x2e, 0x70, 0x65, 0x65, 0x72, 0x64, 0x62, 0x5f, 0x70, 0x65, + 0x65, 0x72, 0x73, 0x2e, 0x45, 0x76, 0x65, 0x6e, 0x74, 0x48, 0x75, 0x62, 0x43, 0x6f, 0x6e, 0x66, + 0x69, 0x67, 0x52, 0x05, 0x76, 0x61, 0x6c, 0x75, 0x65, 0x3a, 0x02, 0x38, 0x01, 0x22, 0xe0, 0x02, + 0x0a, 0x08, 0x53, 0x33, 0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x12, 0x10, 0x0a, 0x03, 0x75, 0x72, + 0x6c, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x03, 0x75, 0x72, 0x6c, 0x12, 0x27, 0x0a, 0x0d, + 0x61, 0x63, 0x63, 0x65, 0x73, 0x73, 0x5f, 0x6b, 0x65, 0x79, 0x5f, 0x69, 0x64, 0x18, 0x02, 0x20, + 0x01, 0x28, 0x09, 0x48, 0x00, 0x52, 0x0b, 0x61, 0x63, 0x63, 0x65, 0x73, 0x73, 0x4b, 0x65, 0x79, + 0x49, 0x64, 0x88, 0x01, 0x01, 0x12, 0x2f, 0x0a, 0x11, 0x73, 0x65, 0x63, 0x72, 0x65, 0x74, 0x5f, + 0x61, 0x63, 0x63, 0x65, 0x73, 0x73, 0x5f, 0x6b, 0x65, 0x79, 0x18, 0x03, 0x20, 0x01, 0x28, 0x09, + 0x48, 0x01, 0x52, 0x0f, 0x73, 0x65, 0x63, 0x72, 0x65, 0x74, 0x41, 0x63, 0x63, 0x65, 0x73, 0x73, + 0x4b, 0x65, 0x79, 0x88, 0x01, 0x01, 0x12, 0x1e, 0x0a, 0x08, 0x72, 0x6f, 0x6c, 0x65, 0x5f, 0x61, + 0x72, 0x6e, 0x18, 0x04, 0x20, 0x01, 0x28, 0x09, 0x48, 0x02, 0x52, 0x07, 0x72, 0x6f, 0x6c, 0x65, + 0x41, 0x72, 0x6e, 0x88, 0x01, 0x01, 0x12, 0x1b, 0x0a, 0x06, 0x72, 0x65, 0x67, 0x69, 0x6f, 0x6e, + 0x18, 0x05, 0x20, 0x01, 0x28, 0x09, 0x48, 0x03, 0x52, 0x06, 0x72, 0x65, 0x67, 0x69, 0x6f, 0x6e, + 0x88, 0x01, 0x01, 0x12, 0x1f, 0x0a, 0x08, 0x65, 0x6e, 0x64, 0x70, 0x6f, 0x69, 0x6e, 0x74, 0x18, + 0x06, 0x20, 0x01, 0x28, 0x09, 0x48, 0x04, 0x52, 0x08, 0x65, 0x6e, 0x64, 0x70, 0x6f, 0x69, 0x6e, + 0x74, 0x88, 0x01, 0x01, 0x12, 0x3d, 0x0a, 0x0b, 0x6d, 0x65, 0x74, 0x61, 0x64, 0x61, 0x74, 0x61, + 0x5f, 0x64, 0x62, 0x18, 0x07, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x1c, 0x2e, 0x70, 0x65, 0x65, 0x72, + 0x64, 0x62, 0x5f, 0x70, 0x65, 0x65, 0x72, 0x73, 0x2e, 0x50, 0x6f, 0x73, 0x74, 0x67, 0x72, 0x65, + 0x73, 0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x52, 0x0a, 0x6d, 0x65, 0x74, 0x61, 0x64, 0x61, 0x74, + 0x61, 0x44, 0x62, 0x42, 0x10, 0x0a, 0x0e, 0x5f, 0x61, 0x63, 0x63, 0x65, 0x73, 0x73, 0x5f, 0x6b, + 0x65, 0x79, 0x5f, 0x69, 0x64, 0x42, 0x14, 0x0a, 0x12, 0x5f, 0x73, 0x65, 0x63, 0x72, 0x65, 0x74, + 0x5f, 0x61, 0x63, 0x63, 0x65, 0x73, 0x73, 0x5f, 0x6b, 0x65, 0x79, 0x42, 0x0b, 0x0a, 0x09, 0x5f, + 0x72, 0x6f, 0x6c, 0x65, 0x5f, 0x61, 0x72, 0x6e, 0x42, 0x09, 0x0a, 0x07, 0x5f, 0x72, 0x65, 0x67, + 0x69, 0x6f, 0x6e, 0x42, 0x0b, 0x0a, 0x09, 0x5f, 0x65, 0x6e, 0x64, 0x70, 0x6f, 0x69, 0x6e, 0x74, + 0x22, 0x89, 0x01, 0x0a, 0x0f, 0x53, 0x71, 0x6c, 0x53, 0x65, 0x72, 0x76, 0x65, 0x72, 0x43, 0x6f, + 0x6e, 0x66, 0x69, 0x67, 0x12, 0x16, 0x0a, 0x06, 0x73, 0x65, 0x72, 0x76, 0x65, 0x72, 0x18, 0x01, + 0x20, 0x01, 0x28, 0x09, 0x52, 0x06, 0x73, 0x65, 0x72, 0x76, 0x65, 0x72, 0x12, 0x12, 0x0a, 0x04, 0x70, 0x6f, 0x72, 0x74, 0x18, 0x02, 0x20, 0x01, 0x28, 0x0d, 0x52, 0x04, 0x70, 0x6f, 0x72, 0x74, 0x12, 0x12, 0x0a, 0x04, 0x75, 0x73, 0x65, 0x72, 0x18, 0x03, 0x20, 0x01, 0x28, 0x09, 0x52, 0x04, 0x75, 0x73, 0x65, 0x72, 0x12, 0x1a, 0x0a, 0x08, 0x70, 0x61, 0x73, 0x73, 0x77, 0x6f, 0x72, 0x64, 0x18, 0x04, 0x20, 0x01, 0x28, 0x09, 0x52, 0x08, 0x70, 0x61, 0x73, 0x73, 0x77, 0x6f, 0x72, 0x64, 0x12, 0x1a, 0x0a, 0x08, 0x64, 0x61, 0x74, 0x61, 0x62, 0x61, 0x73, 0x65, 0x18, 0x05, 0x20, 0x01, - 0x28, 0x09, 0x52, 0x08, 0x64, 0x61, 0x74, 0x61, 0x62, 0x61, 0x73, 0x65, 0x12, 0x31, 0x0a, 0x14, - 0x74, 0x72, 0x61, 0x6e, 0x73, 0x61, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x5f, 0x73, 0x6e, 0x61, 0x70, - 0x73, 0x68, 0x6f, 0x74, 0x18, 0x06, 0x20, 0x01, 0x28, 0x09, 0x52, 0x13, 0x74, 0x72, 0x61, 0x6e, - 0x73, 0x61, 0x63, 0x74, 0x69, 0x6f, 0x6e, 0x53, 0x6e, 0x61, 0x70, 0x73, 0x68, 0x6f, 0x74, 0x22, - 0xbd, 0x02, 0x0a, 0x0e, 0x45, 0x76, 0x65, 0x6e, 0x74, 0x48, 0x75, 0x62, 0x43, 0x6f, 0x6e, 0x66, - 0x69, 0x67, 0x12, 0x1c, 0x0a, 0x09, 0x6e, 0x61, 0x6d, 0x65, 0x73, 0x70, 0x61, 0x63, 0x65, 0x18, - 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x09, 0x6e, 0x61, 0x6d, 0x65, 0x73, 0x70, 0x61, 0x63, 0x65, - 0x12, 0x25, 0x0a, 0x0e, 0x72, 0x65, 0x73, 0x6f, 0x75, 0x72, 0x63, 0x65, 0x5f, 0x67, 0x72, 0x6f, - 0x75, 0x70, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x0d, 0x72, 0x65, 0x73, 0x6f, 0x75, 0x72, - 0x63, 0x65, 0x47, 0x72, 0x6f, 0x75, 0x70, 0x12, 0x1a, 0x0a, 0x08, 0x6c, 0x6f, 0x63, 0x61, 0x74, - 0x69, 0x6f, 0x6e, 0x18, 0x03, 0x20, 0x01, 0x28, 0x09, 0x52, 0x08, 0x6c, 0x6f, 0x63, 0x61, 0x74, - 0x69, 0x6f, 0x6e, 0x12, 0x3d, 0x0a, 0x0b, 0x6d, 0x65, 0x74, 0x61, 0x64, 0x61, 0x74, 0x61, 0x5f, - 0x64, 0x62, 0x18, 0x04, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x1c, 0x2e, 0x70, 0x65, 0x65, 0x72, 0x64, - 0x62, 0x5f, 0x70, 0x65, 0x65, 0x72, 0x73, 0x2e, 0x50, 0x6f, 0x73, 0x74, 0x67, 0x72, 0x65, 0x73, - 0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x52, 0x0a, 0x6d, 0x65, 0x74, 0x61, 0x64, 0x61, 0x74, 0x61, - 0x44, 0x62, 0x12, 0x27, 0x0a, 0x0f, 0x73, 0x75, 0x62, 0x73, 0x63, 0x72, 0x69, 0x70, 0x74, 0x69, - 0x6f, 0x6e, 0x5f, 0x69, 0x64, 0x18, 0x05, 0x20, 0x01, 0x28, 0x09, 0x52, 0x0e, 0x73, 0x75, 0x62, - 0x73, 0x63, 0x72, 0x69, 0x70, 0x74, 0x69, 0x6f, 0x6e, 0x49, 0x64, 0x12, 0x27, 0x0a, 0x0f, 0x70, - 0x61, 0x72, 0x74, 0x69, 0x74, 0x69, 0x6f, 0x6e, 0x5f, 0x63, 0x6f, 0x75, 0x6e, 0x74, 0x18, 0x06, - 0x20, 0x01, 0x28, 0x0d, 0x52, 0x0e, 0x70, 0x61, 0x72, 0x74, 0x69, 0x74, 0x69, 0x6f, 0x6e, 0x43, - 0x6f, 0x75, 0x6e, 0x74, 0x12, 0x39, 0x0a, 0x19, 0x6d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x5f, - 0x72, 0x65, 0x74, 0x65, 0x6e, 0x74, 0x69, 0x6f, 0x6e, 0x5f, 0x69, 0x6e, 0x5f, 0x64, 0x61, 0x79, - 0x73, 0x18, 0x07, 0x20, 0x01, 0x28, 0x0d, 0x52, 0x16, 0x6d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, - 0x52, 0x65, 0x74, 0x65, 0x6e, 0x74, 0x69, 0x6f, 0x6e, 0x49, 0x6e, 0x44, 0x61, 0x79, 0x73, 0x22, - 0xa7, 0x02, 0x0a, 0x13, 0x45, 0x76, 0x65, 0x6e, 0x74, 0x48, 0x75, 0x62, 0x47, 0x72, 0x6f, 0x75, - 0x70, 0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x12, 0x4e, 0x0a, 0x09, 0x65, 0x76, 0x65, 0x6e, 0x74, - 0x68, 0x75, 0x62, 0x73, 0x18, 0x01, 0x20, 0x03, 0x28, 0x0b, 0x32, 0x30, 0x2e, 0x70, 0x65, 0x65, + 0x28, 0x09, 0x52, 0x08, 0x64, 0x61, 0x74, 0x61, 0x62, 0x61, 0x73, 0x65, 0x22, 0x91, 0x05, 0x0a, + 0x04, 0x50, 0x65, 0x65, 0x72, 0x12, 0x12, 0x0a, 0x04, 0x6e, 0x61, 0x6d, 0x65, 0x18, 0x01, 0x20, + 0x01, 0x28, 0x09, 0x52, 0x04, 0x6e, 0x61, 0x6d, 0x65, 0x12, 0x28, 0x0a, 0x04, 0x74, 0x79, 0x70, + 0x65, 0x18, 0x02, 0x20, 0x01, 0x28, 0x0e, 0x32, 0x14, 0x2e, 0x70, 0x65, 0x65, 0x72, 0x64, 0x62, + 0x5f, 0x70, 0x65, 0x65, 0x72, 0x73, 0x2e, 0x44, 0x42, 0x54, 0x79, 0x70, 0x65, 0x52, 0x04, 0x74, + 0x79, 0x70, 0x65, 0x12, 0x4a, 0x0a, 0x10, 0x73, 0x6e, 0x6f, 0x77, 0x66, 0x6c, 0x61, 0x6b, 0x65, + 0x5f, 0x63, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x18, 0x03, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x1d, 0x2e, + 0x70, 0x65, 0x65, 0x72, 0x64, 0x62, 0x5f, 0x70, 0x65, 0x65, 0x72, 0x73, 0x2e, 0x53, 0x6e, 0x6f, + 0x77, 0x66, 0x6c, 0x61, 0x6b, 0x65, 0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x48, 0x00, 0x52, 0x0f, + 0x73, 0x6e, 0x6f, 0x77, 0x66, 0x6c, 0x61, 0x6b, 0x65, 0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x12, + 0x47, 0x0a, 0x0f, 0x62, 0x69, 0x67, 0x71, 0x75, 0x65, 0x72, 0x79, 0x5f, 0x63, 0x6f, 0x6e, 0x66, + 0x69, 0x67, 0x18, 0x04, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x1c, 0x2e, 0x70, 0x65, 0x65, 0x72, 0x64, + 0x62, 0x5f, 0x70, 0x65, 0x65, 0x72, 0x73, 0x2e, 0x42, 0x69, 0x67, 0x71, 0x75, 0x65, 0x72, 0x79, + 0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x48, 0x00, 0x52, 0x0e, 0x62, 0x69, 0x67, 0x71, 0x75, 0x65, + 0x72, 0x79, 0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x12, 0x3e, 0x0a, 0x0c, 0x6d, 0x6f, 0x6e, 0x67, + 0x6f, 0x5f, 0x63, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x18, 0x05, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x19, + 0x2e, 0x70, 0x65, 0x65, 0x72, 0x64, 0x62, 0x5f, 0x70, 0x65, 0x65, 0x72, 0x73, 0x2e, 0x4d, 0x6f, + 0x6e, 0x67, 0x6f, 0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x48, 0x00, 0x52, 0x0b, 0x6d, 0x6f, 0x6e, + 0x67, 0x6f, 0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x12, 0x47, 0x0a, 0x0f, 0x70, 0x6f, 0x73, 0x74, + 0x67, 0x72, 0x65, 0x73, 0x5f, 0x63, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x18, 0x06, 0x20, 0x01, 0x28, + 0x0b, 0x32, 0x1c, 0x2e, 0x70, 0x65, 0x65, 0x72, 0x64, 0x62, 0x5f, 0x70, 0x65, 0x65, 0x72, 0x73, + 0x2e, 0x50, 0x6f, 0x73, 0x74, 0x67, 0x72, 0x65, 0x73, 0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x48, + 0x00, 0x52, 0x0e, 0x70, 0x6f, 0x73, 0x74, 0x67, 0x72, 0x65, 0x73, 0x43, 0x6f, 0x6e, 0x66, 0x69, + 0x67, 0x12, 0x47, 0x0a, 0x0f, 0x65, 0x76, 0x65, 0x6e, 0x74, 0x68, 0x75, 0x62, 0x5f, 0x63, 0x6f, + 0x6e, 0x66, 0x69, 0x67, 0x18, 0x07, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x1c, 0x2e, 0x70, 0x65, 0x65, 0x72, 0x64, 0x62, 0x5f, 0x70, 0x65, 0x65, 0x72, 0x73, 0x2e, 0x45, 0x76, 0x65, 0x6e, 0x74, 0x48, - 0x75, 0x62, 0x47, 0x72, 0x6f, 0x75, 0x70, 0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x2e, 0x45, 0x76, - 0x65, 0x6e, 0x74, 0x68, 0x75, 0x62, 0x73, 0x45, 0x6e, 0x74, 0x72, 0x79, 0x52, 0x09, 0x65, 0x76, - 0x65, 0x6e, 0x74, 0x68, 0x75, 0x62, 0x73, 0x12, 0x3d, 0x0a, 0x0b, 0x6d, 0x65, 0x74, 0x61, 0x64, - 0x61, 0x74, 0x61, 0x5f, 0x64, 0x62, 0x18, 0x02, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x1c, 0x2e, 0x70, - 0x65, 0x65, 0x72, 0x64, 0x62, 0x5f, 0x70, 0x65, 0x65, 0x72, 0x73, 0x2e, 0x50, 0x6f, 0x73, 0x74, - 0x67, 0x72, 0x65, 0x73, 0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x52, 0x0a, 0x6d, 0x65, 0x74, 0x61, - 0x64, 0x61, 0x74, 0x61, 0x44, 0x62, 0x12, 0x25, 0x0a, 0x0e, 0x75, 0x6e, 0x6e, 0x65, 0x73, 0x74, - 0x5f, 0x63, 0x6f, 0x6c, 0x75, 0x6d, 0x6e, 0x73, 0x18, 0x03, 0x20, 0x03, 0x28, 0x09, 0x52, 0x0d, - 0x75, 0x6e, 0x6e, 0x65, 0x73, 0x74, 0x43, 0x6f, 0x6c, 0x75, 0x6d, 0x6e, 0x73, 0x1a, 0x5a, 0x0a, - 0x0e, 0x45, 0x76, 0x65, 0x6e, 0x74, 0x68, 0x75, 0x62, 0x73, 0x45, 0x6e, 0x74, 0x72, 0x79, 0x12, - 0x10, 0x0a, 0x03, 0x6b, 0x65, 0x79, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x03, 0x6b, 0x65, - 0x79, 0x12, 0x32, 0x0a, 0x05, 0x76, 0x61, 0x6c, 0x75, 0x65, 0x18, 0x02, 0x20, 0x01, 0x28, 0x0b, - 0x32, 0x1c, 0x2e, 0x70, 0x65, 0x65, 0x72, 0x64, 0x62, 0x5f, 0x70, 0x65, 0x65, 0x72, 0x73, 0x2e, - 0x45, 0x76, 0x65, 0x6e, 0x74, 0x48, 0x75, 0x62, 0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x52, 0x05, - 0x76, 0x61, 0x6c, 0x75, 0x65, 0x3a, 0x02, 0x38, 0x01, 0x22, 0xe0, 0x02, 0x0a, 0x08, 0x53, 0x33, - 0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x12, 0x10, 0x0a, 0x03, 0x75, 0x72, 0x6c, 0x18, 0x01, 0x20, - 0x01, 0x28, 0x09, 0x52, 0x03, 0x75, 0x72, 0x6c, 0x12, 0x27, 0x0a, 0x0d, 0x61, 0x63, 0x63, 0x65, - 0x73, 0x73, 0x5f, 0x6b, 0x65, 0x79, 0x5f, 0x69, 0x64, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x48, - 0x00, 0x52, 0x0b, 0x61, 0x63, 0x63, 0x65, 0x73, 0x73, 0x4b, 0x65, 0x79, 0x49, 0x64, 0x88, 0x01, - 0x01, 0x12, 0x2f, 0x0a, 0x11, 0x73, 0x65, 0x63, 0x72, 0x65, 0x74, 0x5f, 0x61, 0x63, 0x63, 0x65, - 0x73, 0x73, 0x5f, 0x6b, 0x65, 0x79, 0x18, 0x03, 0x20, 0x01, 0x28, 0x09, 0x48, 0x01, 0x52, 0x0f, - 0x73, 0x65, 0x63, 0x72, 0x65, 0x74, 0x41, 0x63, 0x63, 0x65, 0x73, 0x73, 0x4b, 0x65, 0x79, 0x88, - 0x01, 0x01, 0x12, 0x1e, 0x0a, 0x08, 0x72, 0x6f, 0x6c, 0x65, 0x5f, 0x61, 0x72, 0x6e, 0x18, 0x04, - 0x20, 0x01, 0x28, 0x09, 0x48, 0x02, 0x52, 0x07, 0x72, 0x6f, 0x6c, 0x65, 0x41, 0x72, 0x6e, 0x88, - 0x01, 0x01, 0x12, 0x1b, 0x0a, 0x06, 0x72, 0x65, 0x67, 0x69, 0x6f, 0x6e, 0x18, 0x05, 0x20, 0x01, - 0x28, 0x09, 0x48, 0x03, 0x52, 0x06, 0x72, 0x65, 0x67, 0x69, 0x6f, 0x6e, 0x88, 0x01, 0x01, 0x12, - 0x1f, 0x0a, 0x08, 0x65, 0x6e, 0x64, 0x70, 0x6f, 0x69, 0x6e, 0x74, 0x18, 0x06, 0x20, 0x01, 0x28, - 0x09, 0x48, 0x04, 0x52, 0x08, 0x65, 0x6e, 0x64, 0x70, 0x6f, 0x69, 0x6e, 0x74, 0x88, 0x01, 0x01, - 0x12, 0x3d, 0x0a, 0x0b, 0x6d, 0x65, 0x74, 0x61, 0x64, 0x61, 0x74, 0x61, 0x5f, 0x64, 0x62, 0x18, - 0x07, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x1c, 0x2e, 0x70, 0x65, 0x65, 0x72, 0x64, 0x62, 0x5f, 0x70, - 0x65, 0x65, 0x72, 0x73, 0x2e, 0x50, 0x6f, 0x73, 0x74, 0x67, 0x72, 0x65, 0x73, 0x43, 0x6f, 0x6e, - 0x66, 0x69, 0x67, 0x52, 0x0a, 0x6d, 0x65, 0x74, 0x61, 0x64, 0x61, 0x74, 0x61, 0x44, 0x62, 0x42, - 0x10, 0x0a, 0x0e, 0x5f, 0x61, 0x63, 0x63, 0x65, 0x73, 0x73, 0x5f, 0x6b, 0x65, 0x79, 0x5f, 0x69, - 0x64, 0x42, 0x14, 0x0a, 0x12, 0x5f, 0x73, 0x65, 0x63, 0x72, 0x65, 0x74, 0x5f, 0x61, 0x63, 0x63, - 0x65, 0x73, 0x73, 0x5f, 0x6b, 0x65, 0x79, 0x42, 0x0b, 0x0a, 0x09, 0x5f, 0x72, 0x6f, 0x6c, 0x65, - 0x5f, 0x61, 0x72, 0x6e, 0x42, 0x09, 0x0a, 0x07, 0x5f, 0x72, 0x65, 0x67, 0x69, 0x6f, 0x6e, 0x42, - 0x0b, 0x0a, 0x09, 0x5f, 0x65, 0x6e, 0x64, 0x70, 0x6f, 0x69, 0x6e, 0x74, 0x22, 0x89, 0x01, 0x0a, - 0x0f, 0x53, 0x71, 0x6c, 0x53, 0x65, 0x72, 0x76, 0x65, 0x72, 0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67, - 0x12, 0x16, 0x0a, 0x06, 0x73, 0x65, 0x72, 0x76, 0x65, 0x72, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, - 0x52, 0x06, 0x73, 0x65, 0x72, 0x76, 0x65, 0x72, 0x12, 0x12, 0x0a, 0x04, 0x70, 0x6f, 0x72, 0x74, - 0x18, 0x02, 0x20, 0x01, 0x28, 0x0d, 0x52, 0x04, 0x70, 0x6f, 0x72, 0x74, 0x12, 0x12, 0x0a, 0x04, - 0x75, 0x73, 0x65, 0x72, 0x18, 0x03, 0x20, 0x01, 0x28, 0x09, 0x52, 0x04, 0x75, 0x73, 0x65, 0x72, - 0x12, 0x1a, 0x0a, 0x08, 0x70, 0x61, 0x73, 0x73, 0x77, 0x6f, 0x72, 0x64, 0x18, 0x04, 0x20, 0x01, - 0x28, 0x09, 0x52, 0x08, 0x70, 0x61, 0x73, 0x73, 0x77, 0x6f, 0x72, 0x64, 0x12, 0x1a, 0x0a, 0x08, - 0x64, 0x61, 0x74, 0x61, 0x62, 0x61, 0x73, 0x65, 0x18, 0x05, 0x20, 0x01, 0x28, 0x09, 0x52, 0x08, - 0x64, 0x61, 0x74, 0x61, 0x62, 0x61, 0x73, 0x65, 0x22, 0x91, 0x05, 0x0a, 0x04, 0x50, 0x65, 0x65, - 0x72, 0x12, 0x12, 0x0a, 0x04, 0x6e, 0x61, 0x6d, 0x65, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, - 0x04, 0x6e, 0x61, 0x6d, 0x65, 0x12, 0x28, 0x0a, 0x04, 0x74, 0x79, 0x70, 0x65, 0x18, 0x02, 0x20, - 0x01, 0x28, 0x0e, 0x32, 0x14, 0x2e, 0x70, 0x65, 0x65, 0x72, 0x64, 0x62, 0x5f, 0x70, 0x65, 0x65, - 0x72, 0x73, 0x2e, 0x44, 0x42, 0x54, 0x79, 0x70, 0x65, 0x52, 0x04, 0x74, 0x79, 0x70, 0x65, 0x12, - 0x4a, 0x0a, 0x10, 0x73, 0x6e, 0x6f, 0x77, 0x66, 0x6c, 0x61, 0x6b, 0x65, 0x5f, 0x63, 0x6f, 0x6e, - 0x66, 0x69, 0x67, 0x18, 0x03, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x1d, 0x2e, 0x70, 0x65, 0x65, 0x72, - 0x64, 0x62, 0x5f, 0x70, 0x65, 0x65, 0x72, 0x73, 0x2e, 0x53, 0x6e, 0x6f, 0x77, 0x66, 0x6c, 0x61, - 0x6b, 0x65, 0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x48, 0x00, 0x52, 0x0f, 0x73, 0x6e, 0x6f, 0x77, - 0x66, 0x6c, 0x61, 0x6b, 0x65, 0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x12, 0x47, 0x0a, 0x0f, 0x62, - 0x69, 0x67, 0x71, 0x75, 0x65, 0x72, 0x79, 0x5f, 0x63, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x18, 0x04, - 0x20, 0x01, 0x28, 0x0b, 0x32, 0x1c, 0x2e, 0x70, 0x65, 0x65, 0x72, 0x64, 0x62, 0x5f, 0x70, 0x65, - 0x65, 0x72, 0x73, 0x2e, 0x42, 0x69, 0x67, 0x71, 0x75, 0x65, 0x72, 0x79, 0x43, 0x6f, 0x6e, 0x66, - 0x69, 0x67, 0x48, 0x00, 0x52, 0x0e, 0x62, 0x69, 0x67, 0x71, 0x75, 0x65, 0x72, 0x79, 0x43, 0x6f, - 0x6e, 0x66, 0x69, 0x67, 0x12, 0x3e, 0x0a, 0x0c, 0x6d, 0x6f, 0x6e, 0x67, 0x6f, 0x5f, 0x63, 0x6f, - 0x6e, 0x66, 0x69, 0x67, 0x18, 0x05, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x19, 0x2e, 0x70, 0x65, 0x65, - 0x72, 0x64, 0x62, 0x5f, 0x70, 0x65, 0x65, 0x72, 0x73, 0x2e, 0x4d, 0x6f, 0x6e, 0x67, 0x6f, 0x43, - 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x48, 0x00, 0x52, 0x0b, 0x6d, 0x6f, 0x6e, 0x67, 0x6f, 0x43, 0x6f, - 0x6e, 0x66, 0x69, 0x67, 0x12, 0x47, 0x0a, 0x0f, 0x70, 0x6f, 0x73, 0x74, 0x67, 0x72, 0x65, 0x73, - 0x5f, 0x63, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x18, 0x06, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x1c, 0x2e, - 0x70, 0x65, 0x65, 0x72, 0x64, 0x62, 0x5f, 0x70, 0x65, 0x65, 0x72, 0x73, 0x2e, 0x50, 0x6f, 0x73, - 0x74, 0x67, 0x72, 0x65, 0x73, 0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x48, 0x00, 0x52, 0x0e, 0x70, - 0x6f, 0x73, 0x74, 0x67, 0x72, 0x65, 0x73, 0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x12, 0x47, 0x0a, - 0x0f, 0x65, 0x76, 0x65, 0x6e, 0x74, 0x68, 0x75, 0x62, 0x5f, 0x63, 0x6f, 0x6e, 0x66, 0x69, 0x67, - 0x18, 0x07, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x1c, 0x2e, 0x70, 0x65, 0x65, 0x72, 0x64, 0x62, 0x5f, - 0x70, 0x65, 0x65, 0x72, 0x73, 0x2e, 0x45, 0x76, 0x65, 0x6e, 0x74, 0x48, 0x75, 0x62, 0x43, 0x6f, - 0x6e, 0x66, 0x69, 0x67, 0x48, 0x00, 0x52, 0x0e, 0x65, 0x76, 0x65, 0x6e, 0x74, 0x68, 0x75, 0x62, - 0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x12, 0x35, 0x0a, 0x09, 0x73, 0x33, 0x5f, 0x63, 0x6f, 0x6e, - 0x66, 0x69, 0x67, 0x18, 0x08, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x16, 0x2e, 0x70, 0x65, 0x65, 0x72, - 0x64, 0x62, 0x5f, 0x70, 0x65, 0x65, 0x72, 0x73, 0x2e, 0x53, 0x33, 0x43, 0x6f, 0x6e, 0x66, 0x69, - 0x67, 0x48, 0x00, 0x52, 0x08, 0x73, 0x33, 0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x12, 0x4a, 0x0a, - 0x10, 0x73, 0x71, 0x6c, 0x73, 0x65, 0x72, 0x76, 0x65, 0x72, 0x5f, 0x63, 0x6f, 0x6e, 0x66, 0x69, - 0x67, 0x18, 0x09, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x1d, 0x2e, 0x70, 0x65, 0x65, 0x72, 0x64, 0x62, - 0x5f, 0x70, 0x65, 0x65, 0x72, 0x73, 0x2e, 0x53, 0x71, 0x6c, 0x53, 0x65, 0x72, 0x76, 0x65, 0x72, - 0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x48, 0x00, 0x52, 0x0f, 0x73, 0x71, 0x6c, 0x73, 0x65, 0x72, - 0x76, 0x65, 0x72, 0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x12, 0x57, 0x0a, 0x15, 0x65, 0x76, 0x65, - 0x6e, 0x74, 0x68, 0x75, 0x62, 0x5f, 0x67, 0x72, 0x6f, 0x75, 0x70, 0x5f, 0x63, 0x6f, 0x6e, 0x66, - 0x69, 0x67, 0x18, 0x0a, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x21, 0x2e, 0x70, 0x65, 0x65, 0x72, 0x64, - 0x62, 0x5f, 0x70, 0x65, 0x65, 0x72, 0x73, 0x2e, 0x45, 0x76, 0x65, 0x6e, 0x74, 0x48, 0x75, 0x62, - 0x47, 0x72, 0x6f, 0x75, 0x70, 0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x48, 0x00, 0x52, 0x13, 0x65, - 0x76, 0x65, 0x6e, 0x74, 0x68, 0x75, 0x62, 0x47, 0x72, 0x6f, 0x75, 0x70, 0x43, 0x6f, 0x6e, 0x66, - 0x69, 0x67, 0x42, 0x08, 0x0a, 0x06, 0x63, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x2a, 0x77, 0x0a, 0x06, - 0x44, 0x42, 0x54, 0x79, 0x70, 0x65, 0x12, 0x0c, 0x0a, 0x08, 0x42, 0x49, 0x47, 0x51, 0x55, 0x45, - 0x52, 0x59, 0x10, 0x00, 0x12, 0x0d, 0x0a, 0x09, 0x53, 0x4e, 0x4f, 0x57, 0x46, 0x4c, 0x41, 0x4b, - 0x45, 0x10, 0x01, 0x12, 0x09, 0x0a, 0x05, 0x4d, 0x4f, 0x4e, 0x47, 0x4f, 0x10, 0x02, 0x12, 0x0c, - 0x0a, 0x08, 0x50, 0x4f, 0x53, 0x54, 0x47, 0x52, 0x45, 0x53, 0x10, 0x03, 0x12, 0x0c, 0x0a, 0x08, - 0x45, 0x56, 0x45, 0x4e, 0x54, 0x48, 0x55, 0x42, 0x10, 0x04, 0x12, 0x06, 0x0a, 0x02, 0x53, 0x33, - 0x10, 0x05, 0x12, 0x0d, 0x0a, 0x09, 0x53, 0x51, 0x4c, 0x53, 0x45, 0x52, 0x56, 0x45, 0x52, 0x10, - 0x06, 0x12, 0x12, 0x0a, 0x0e, 0x45, 0x56, 0x45, 0x4e, 0x54, 0x48, 0x55, 0x42, 0x5f, 0x47, 0x52, - 0x4f, 0x55, 0x50, 0x10, 0x07, 0x42, 0x7c, 0x0a, 0x10, 0x63, 0x6f, 0x6d, 0x2e, 0x70, 0x65, 0x65, - 0x72, 0x64, 0x62, 0x5f, 0x70, 0x65, 0x65, 0x72, 0x73, 0x42, 0x0a, 0x50, 0x65, 0x65, 0x72, 0x73, - 0x50, 0x72, 0x6f, 0x74, 0x6f, 0x50, 0x01, 0x5a, 0x10, 0x67, 0x65, 0x6e, 0x65, 0x72, 0x61, 0x74, - 0x65, 0x64, 0x2f, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x73, 0xa2, 0x02, 0x03, 0x50, 0x58, 0x58, 0xaa, - 0x02, 0x0b, 0x50, 0x65, 0x65, 0x72, 0x64, 0x62, 0x50, 0x65, 0x65, 0x72, 0x73, 0xca, 0x02, 0x0b, - 0x50, 0x65, 0x65, 0x72, 0x64, 0x62, 0x50, 0x65, 0x65, 0x72, 0x73, 0xe2, 0x02, 0x17, 0x50, 0x65, - 0x65, 0x72, 0x64, 0x62, 0x50, 0x65, 0x65, 0x72, 0x73, 0x5c, 0x47, 0x50, 0x42, 0x4d, 0x65, 0x74, - 0x61, 0x64, 0x61, 0x74, 0x61, 0xea, 0x02, 0x0b, 0x50, 0x65, 0x65, 0x72, 0x64, 0x62, 0x50, 0x65, - 0x65, 0x72, 0x73, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33, + 0x75, 0x62, 0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x48, 0x00, 0x52, 0x0e, 0x65, 0x76, 0x65, 0x6e, + 0x74, 0x68, 0x75, 0x62, 0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x12, 0x35, 0x0a, 0x09, 0x73, 0x33, + 0x5f, 0x63, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x18, 0x08, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x16, 0x2e, + 0x70, 0x65, 0x65, 0x72, 0x64, 0x62, 0x5f, 0x70, 0x65, 0x65, 0x72, 0x73, 0x2e, 0x53, 0x33, 0x43, + 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x48, 0x00, 0x52, 0x08, 0x73, 0x33, 0x43, 0x6f, 0x6e, 0x66, 0x69, + 0x67, 0x12, 0x4a, 0x0a, 0x10, 0x73, 0x71, 0x6c, 0x73, 0x65, 0x72, 0x76, 0x65, 0x72, 0x5f, 0x63, + 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x18, 0x09, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x1d, 0x2e, 0x70, 0x65, + 0x65, 0x72, 0x64, 0x62, 0x5f, 0x70, 0x65, 0x65, 0x72, 0x73, 0x2e, 0x53, 0x71, 0x6c, 0x53, 0x65, + 0x72, 0x76, 0x65, 0x72, 0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x48, 0x00, 0x52, 0x0f, 0x73, 0x71, + 0x6c, 0x73, 0x65, 0x72, 0x76, 0x65, 0x72, 0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x12, 0x57, 0x0a, + 0x15, 0x65, 0x76, 0x65, 0x6e, 0x74, 0x68, 0x75, 0x62, 0x5f, 0x67, 0x72, 0x6f, 0x75, 0x70, 0x5f, + 0x63, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x18, 0x0a, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x21, 0x2e, 0x70, + 0x65, 0x65, 0x72, 0x64, 0x62, 0x5f, 0x70, 0x65, 0x65, 0x72, 0x73, 0x2e, 0x45, 0x76, 0x65, 0x6e, + 0x74, 0x48, 0x75, 0x62, 0x47, 0x72, 0x6f, 0x75, 0x70, 0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x48, + 0x00, 0x52, 0x13, 0x65, 0x76, 0x65, 0x6e, 0x74, 0x68, 0x75, 0x62, 0x47, 0x72, 0x6f, 0x75, 0x70, + 0x43, 0x6f, 0x6e, 0x66, 0x69, 0x67, 0x42, 0x08, 0x0a, 0x06, 0x63, 0x6f, 0x6e, 0x66, 0x69, 0x67, + 0x2a, 0x77, 0x0a, 0x06, 0x44, 0x42, 0x54, 0x79, 0x70, 0x65, 0x12, 0x0c, 0x0a, 0x08, 0x42, 0x49, + 0x47, 0x51, 0x55, 0x45, 0x52, 0x59, 0x10, 0x00, 0x12, 0x0d, 0x0a, 0x09, 0x53, 0x4e, 0x4f, 0x57, + 0x46, 0x4c, 0x41, 0x4b, 0x45, 0x10, 0x01, 0x12, 0x09, 0x0a, 0x05, 0x4d, 0x4f, 0x4e, 0x47, 0x4f, + 0x10, 0x02, 0x12, 0x0c, 0x0a, 0x08, 0x50, 0x4f, 0x53, 0x54, 0x47, 0x52, 0x45, 0x53, 0x10, 0x03, + 0x12, 0x0c, 0x0a, 0x08, 0x45, 0x56, 0x45, 0x4e, 0x54, 0x48, 0x55, 0x42, 0x10, 0x04, 0x12, 0x06, + 0x0a, 0x02, 0x53, 0x33, 0x10, 0x05, 0x12, 0x0d, 0x0a, 0x09, 0x53, 0x51, 0x4c, 0x53, 0x45, 0x52, + 0x56, 0x45, 0x52, 0x10, 0x06, 0x12, 0x12, 0x0a, 0x0e, 0x45, 0x56, 0x45, 0x4e, 0x54, 0x48, 0x55, + 0x42, 0x5f, 0x47, 0x52, 0x4f, 0x55, 0x50, 0x10, 0x07, 0x42, 0x7c, 0x0a, 0x10, 0x63, 0x6f, 0x6d, + 0x2e, 0x70, 0x65, 0x65, 0x72, 0x64, 0x62, 0x5f, 0x70, 0x65, 0x65, 0x72, 0x73, 0x42, 0x0a, 0x50, + 0x65, 0x65, 0x72, 0x73, 0x50, 0x72, 0x6f, 0x74, 0x6f, 0x50, 0x01, 0x5a, 0x10, 0x67, 0x65, 0x6e, + 0x65, 0x72, 0x61, 0x74, 0x65, 0x64, 0x2f, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x73, 0xa2, 0x02, 0x03, + 0x50, 0x58, 0x58, 0xaa, 0x02, 0x0b, 0x50, 0x65, 0x65, 0x72, 0x64, 0x62, 0x50, 0x65, 0x65, 0x72, + 0x73, 0xca, 0x02, 0x0b, 0x50, 0x65, 0x65, 0x72, 0x64, 0x62, 0x50, 0x65, 0x65, 0x72, 0x73, 0xe2, + 0x02, 0x17, 0x50, 0x65, 0x65, 0x72, 0x64, 0x62, 0x50, 0x65, 0x65, 0x72, 0x73, 0x5c, 0x47, 0x50, + 0x42, 0x4d, 0x65, 0x74, 0x61, 0x64, 0x61, 0x74, 0x61, 0xea, 0x02, 0x0b, 0x50, 0x65, 0x65, 0x72, + 0x64, 0x62, 0x50, 0x65, 0x65, 0x72, 0x73, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33, } var ( @@ -1372,6 +1398,7 @@ func file_peers_proto_init() { } } file_peers_proto_msgTypes[0].OneofWrappers = []interface{}{} + file_peers_proto_msgTypes[3].OneofWrappers = []interface{}{} file_peers_proto_msgTypes[6].OneofWrappers = []interface{}{} file_peers_proto_msgTypes[8].OneofWrappers = []interface{}{ (*Peer_SnowflakeConfig)(nil), diff --git a/nexus/analyzer/src/lib.rs b/nexus/analyzer/src/lib.rs index 6939b70809..14ff6b285a 100644 --- a/nexus/analyzer/src/lib.rs +++ b/nexus/analyzer/src/lib.rs @@ -533,6 +533,7 @@ fn parse_db_options( .parse::() .context("unable to parse query_timeout")?, password: opts.get("password").map(|s| s.to_string()), + metadata_schema: opts.get("metadata_schema").map(|s| s.to_string()), s3_integration: s3_int, }; let config = Config::SnowflakeConfig(snowflake_config); @@ -585,6 +586,7 @@ fn parse_db_options( .get("database") .context("no default database specified")? .to_string(), + metadata_schema: opts.get("metadata_schema").map(|s| s.to_string()), transaction_snapshot: "".to_string(), }; let config = Config::PostgresConfig(postgres_config); diff --git a/nexus/catalog/src/lib.rs b/nexus/catalog/src/lib.rs index 1f751d6ad7..bf00addf83 100644 --- a/nexus/catalog/src/lib.rs +++ b/nexus/catalog/src/lib.rs @@ -73,6 +73,7 @@ impl CatalogConfig { password: self.password.clone(), database: self.database.clone(), transaction_snapshot: "".to_string(), + metadata_schema: Some("".to_string()) } } diff --git a/nexus/pt/src/peerdb_peers.rs b/nexus/pt/src/peerdb_peers.rs index f8a304b146..017a36b8ab 100644 --- a/nexus/pt/src/peerdb_peers.rs +++ b/nexus/pt/src/peerdb_peers.rs @@ -20,6 +20,9 @@ pub struct SnowflakeConfig { pub s3_integration: ::prost::alloc::string::String, #[prost(string, optional, tag="10")] pub password: ::core::option::Option<::prost::alloc::string::String>, + /// defaults to _PEERDB_INTERNAL + #[prost(string, optional, tag="11")] + pub metadata_schema: ::core::option::Option<::prost::alloc::string::String>, } #[allow(clippy::derive_partial_eq_without_eq)] #[derive(Clone, PartialEq, ::prost::Message)] @@ -77,6 +80,9 @@ pub struct PostgresConfig { /// this is used only in query replication mode right now. #[prost(string, tag="6")] pub transaction_snapshot: ::prost::alloc::string::String, + /// defaults to _peerdb_internal + #[prost(string, optional, tag="7")] + pub metadata_schema: ::core::option::Option<::prost::alloc::string::String>, } #[allow(clippy::derive_partial_eq_without_eq)] #[derive(Clone, PartialEq, ::prost::Message)] diff --git a/nexus/pt/src/peerdb_peers.serde.rs b/nexus/pt/src/peerdb_peers.serde.rs index e03dc494ee..2cdf144f4b 100644 --- a/nexus/pt/src/peerdb_peers.serde.rs +++ b/nexus/pt/src/peerdb_peers.serde.rs @@ -1131,6 +1131,9 @@ impl serde::Serialize for PostgresConfig { if !self.transaction_snapshot.is_empty() { len += 1; } + if self.metadata_schema.is_some() { + len += 1; + } let mut struct_ser = serializer.serialize_struct("peerdb_peers.PostgresConfig", len)?; if !self.host.is_empty() { struct_ser.serialize_field("host", &self.host)?; @@ -1150,6 +1153,9 @@ impl serde::Serialize for PostgresConfig { if !self.transaction_snapshot.is_empty() { struct_ser.serialize_field("transactionSnapshot", &self.transaction_snapshot)?; } + if let Some(v) = self.metadata_schema.as_ref() { + struct_ser.serialize_field("metadataSchema", v)?; + } struct_ser.end() } } @@ -1167,6 +1173,8 @@ impl<'de> serde::Deserialize<'de> for PostgresConfig { "database", "transaction_snapshot", "transactionSnapshot", + "metadata_schema", + "metadataSchema", ]; #[allow(clippy::enum_variant_names)] @@ -1177,6 +1185,7 @@ impl<'de> serde::Deserialize<'de> for PostgresConfig { Password, Database, TransactionSnapshot, + MetadataSchema, __SkipField__, } impl<'de> serde::Deserialize<'de> for GeneratedField { @@ -1205,6 +1214,7 @@ impl<'de> serde::Deserialize<'de> for PostgresConfig { "password" => Ok(GeneratedField::Password), "database" => Ok(GeneratedField::Database), "transactionSnapshot" | "transaction_snapshot" => Ok(GeneratedField::TransactionSnapshot), + "metadataSchema" | "metadata_schema" => Ok(GeneratedField::MetadataSchema), _ => Ok(GeneratedField::__SkipField__), } } @@ -1230,6 +1240,7 @@ impl<'de> serde::Deserialize<'de> for PostgresConfig { let mut password__ = None; let mut database__ = None; let mut transaction_snapshot__ = None; + let mut metadata_schema__ = None; while let Some(k) = map.next_key()? { match k { GeneratedField::Host => { @@ -1270,6 +1281,12 @@ impl<'de> serde::Deserialize<'de> for PostgresConfig { } transaction_snapshot__ = Some(map.next_value()?); } + GeneratedField::MetadataSchema => { + if metadata_schema__.is_some() { + return Err(serde::de::Error::duplicate_field("metadataSchema")); + } + metadata_schema__ = map.next_value()?; + } GeneratedField::__SkipField__ => { let _ = map.next_value::()?; } @@ -1282,6 +1299,7 @@ impl<'de> serde::Deserialize<'de> for PostgresConfig { password: password__.unwrap_or_default(), database: database__.unwrap_or_default(), transaction_snapshot: transaction_snapshot__.unwrap_or_default(), + metadata_schema: metadata_schema__, }) } } @@ -1524,6 +1542,9 @@ impl serde::Serialize for SnowflakeConfig { if self.password.is_some() { len += 1; } + if self.metadata_schema.is_some() { + len += 1; + } let mut struct_ser = serializer.serialize_struct("peerdb_peers.SnowflakeConfig", len)?; if !self.account_id.is_empty() { struct_ser.serialize_field("accountId", &self.account_id)?; @@ -1552,6 +1573,9 @@ impl serde::Serialize for SnowflakeConfig { if let Some(v) = self.password.as_ref() { struct_ser.serialize_field("password", v)?; } + if let Some(v) = self.metadata_schema.as_ref() { + struct_ser.serialize_field("metadataSchema", v)?; + } struct_ser.end() } } @@ -1575,6 +1599,8 @@ impl<'de> serde::Deserialize<'de> for SnowflakeConfig { "s3_integration", "s3Integration", "password", + "metadata_schema", + "metadataSchema", ]; #[allow(clippy::enum_variant_names)] @@ -1588,6 +1614,7 @@ impl<'de> serde::Deserialize<'de> for SnowflakeConfig { QueryTimeout, S3Integration, Password, + MetadataSchema, __SkipField__, } impl<'de> serde::Deserialize<'de> for GeneratedField { @@ -1619,6 +1646,7 @@ impl<'de> serde::Deserialize<'de> for SnowflakeConfig { "queryTimeout" | "query_timeout" => Ok(GeneratedField::QueryTimeout), "s3Integration" | "s3_integration" => Ok(GeneratedField::S3Integration), "password" => Ok(GeneratedField::Password), + "metadataSchema" | "metadata_schema" => Ok(GeneratedField::MetadataSchema), _ => Ok(GeneratedField::__SkipField__), } } @@ -1647,6 +1675,7 @@ impl<'de> serde::Deserialize<'de> for SnowflakeConfig { let mut query_timeout__ = None; let mut s3_integration__ = None; let mut password__ = None; + let mut metadata_schema__ = None; while let Some(k) = map.next_key()? { match k { GeneratedField::AccountId => { @@ -1705,6 +1734,12 @@ impl<'de> serde::Deserialize<'de> for SnowflakeConfig { } password__ = map.next_value()?; } + GeneratedField::MetadataSchema => { + if metadata_schema__.is_some() { + return Err(serde::de::Error::duplicate_field("metadataSchema")); + } + metadata_schema__ = map.next_value()?; + } GeneratedField::__SkipField__ => { let _ = map.next_value::()?; } @@ -1720,6 +1755,7 @@ impl<'de> serde::Deserialize<'de> for SnowflakeConfig { query_timeout: query_timeout__.unwrap_or_default(), s3_integration: s3_integration__.unwrap_or_default(), password: password__, + metadata_schema: metadata_schema__, }) } } diff --git a/protos/peers.proto b/protos/peers.proto index 162a63f3ac..fe343d906e 100644 --- a/protos/peers.proto +++ b/protos/peers.proto @@ -12,6 +12,8 @@ message SnowflakeConfig { uint64 query_timeout = 8; string s3_integration = 9; optional string password = 10; + // defaults to _PEERDB_INTERNAL + optional string metadata_schema = 11; } message BigqueryConfig { @@ -42,9 +44,10 @@ message PostgresConfig { string user = 3; string password = 4; string database = 5; - // this is used only in query replication mode right now. string transaction_snapshot = 6; + // defaults to _peerdb_internal + optional string metadata_schema = 7; } message EventHubConfig { diff --git a/ui/grpc_generated/peers.ts b/ui/grpc_generated/peers.ts index 6d9bb4b30d..8d4b9b4eef 100644 --- a/ui/grpc_generated/peers.ts +++ b/ui/grpc_generated/peers.ts @@ -82,7 +82,11 @@ export interface SnowflakeConfig { role: string; queryTimeout: number; s3Integration: string; - password?: string | undefined; + password?: + | string + | undefined; + /** defaults to _PEERDB_INTERNAL */ + metadataSchema?: string | undefined; } export interface BigqueryConfig { @@ -115,6 +119,8 @@ export interface PostgresConfig { database: string; /** this is used only in query replication mode right now. */ transactionSnapshot: string; + /** defaults to _peerdb_internal */ + metadataSchema?: string | undefined; } export interface EventHubConfig { @@ -186,6 +192,7 @@ function createBaseSnowflakeConfig(): SnowflakeConfig { queryTimeout: 0, s3Integration: "", password: undefined, + metadataSchema: undefined, }; } @@ -218,6 +225,9 @@ export const SnowflakeConfig = { if (message.password !== undefined) { writer.uint32(82).string(message.password); } + if (message.metadataSchema !== undefined) { + writer.uint32(90).string(message.metadataSchema); + } return writer; }, @@ -291,6 +301,13 @@ export const SnowflakeConfig = { message.password = reader.string(); continue; + case 11: + if (tag !== 90) { + break; + } + + message.metadataSchema = reader.string(); + continue; } if ((tag & 7) === 4 || tag === 0) { break; @@ -311,6 +328,7 @@ export const SnowflakeConfig = { queryTimeout: isSet(object.queryTimeout) ? Number(object.queryTimeout) : 0, s3Integration: isSet(object.s3Integration) ? String(object.s3Integration) : "", password: isSet(object.password) ? String(object.password) : undefined, + metadataSchema: isSet(object.metadataSchema) ? String(object.metadataSchema) : undefined, }; }, @@ -343,6 +361,9 @@ export const SnowflakeConfig = { if (message.password !== undefined) { obj.password = message.password; } + if (message.metadataSchema !== undefined) { + obj.metadataSchema = message.metadataSchema; + } return obj; }, @@ -360,6 +381,7 @@ export const SnowflakeConfig = { message.queryTimeout = object.queryTimeout ?? 0; message.s3Integration = object.s3Integration ?? ""; message.password = object.password ?? undefined; + message.metadataSchema = object.metadataSchema ?? undefined; return message; }, }; @@ -705,7 +727,15 @@ export const MongoConfig = { }; function createBasePostgresConfig(): PostgresConfig { - return { host: "", port: 0, user: "", password: "", database: "", transactionSnapshot: "" }; + return { + host: "", + port: 0, + user: "", + password: "", + database: "", + transactionSnapshot: "", + metadataSchema: undefined, + }; } export const PostgresConfig = { @@ -728,6 +758,9 @@ export const PostgresConfig = { if (message.transactionSnapshot !== "") { writer.uint32(50).string(message.transactionSnapshot); } + if (message.metadataSchema !== undefined) { + writer.uint32(58).string(message.metadataSchema); + } return writer; }, @@ -780,6 +813,13 @@ export const PostgresConfig = { message.transactionSnapshot = reader.string(); continue; + case 7: + if (tag !== 58) { + break; + } + + message.metadataSchema = reader.string(); + continue; } if ((tag & 7) === 4 || tag === 0) { break; @@ -797,6 +837,7 @@ export const PostgresConfig = { password: isSet(object.password) ? String(object.password) : "", database: isSet(object.database) ? String(object.database) : "", transactionSnapshot: isSet(object.transactionSnapshot) ? String(object.transactionSnapshot) : "", + metadataSchema: isSet(object.metadataSchema) ? String(object.metadataSchema) : undefined, }; }, @@ -820,6 +861,9 @@ export const PostgresConfig = { if (message.transactionSnapshot !== "") { obj.transactionSnapshot = message.transactionSnapshot; } + if (message.metadataSchema !== undefined) { + obj.metadataSchema = message.metadataSchema; + } return obj; }, @@ -834,6 +878,7 @@ export const PostgresConfig = { message.password = object.password ?? ""; message.database = object.database ?? ""; message.transactionSnapshot = object.transactionSnapshot ?? ""; + message.metadataSchema = object.metadataSchema ?? undefined; return message; }, };