cockroach: github.com/cockroachdb/cockroach/pkg/sql Index | Files | Directories

package sql

import "github.com/cockroachdb/cockroach/pkg/sql"

Package sql provides the user-facing API for access to a Cockroach datastore. As the name suggests, the API is based around SQL, the same SQL you find in traditional RDBMS systems like Oracle, MySQL or Postgres. The core Cockroach system implements a distributed, transactional, monolithic sorted key-value map. The sql package builds on top of this core system (provided by the storage and kv packages) adding parsing, query planning and query execution as well as defining the privilege model.

Databases and Tables

The two primary objects are databases and tables. A database is a namespace which holds a series of tables. Conceptually, a database can be viewed as a directory in a filesystem plus some additional metadata. A table is like a file on steroids: it contains a structured layout of rows and columns along with secondary indexes.

Like a directory, a database has a name and some metadata. The metadata is defined by the DatabaseDescriptor:

message DatabaseDescriptor {
  optional string name;
  optional uint32 id;
  optional PrivilegeDescriptor privileges;
}

As you can see, currently the metadata we store for databases just consists of privileges.

Similarly, tables have a TableDescriptor:

message TableDescriptor {
  optional string name;
  optional uint32 id;
  repeated ColumnDescriptor columns;
  optional IndexDescriptor primary_index;
  repeated IndexDescriptor indexes;
  optional PrivilegeDescriptor privileges;
}

Both the database ID and the table ID are allocated from the same "ID space" and IDs are never reused.

The namespace in which databases and tables exist contains only two levels: the root level contains databases and the database level contains tables. The "system.namespace" and "system.descriptor" tables implement the mapping from database/table name to ID and from ID to descriptor:

CREATE TABLE system.namespace (
  "parentID" INT,
  "name"     CHAR,
  "id"       INT,
  PRIMARY KEY ("parentID", name)
);

Create TABLE system.descriptor (
  "id"         INT PRIMARY KEY,
  "descriptor" BLOB
);

The ID 0 is a reserved ID used for the "root" of the namespace in which the databases reside. In order to look up the ID of a database given its name, the system runs the underlying key-value operations that correspond to the following query:

SELECT id FROM system.namespace WHERE "parentID" = 0 AND name = <database-name>

And given a database/table ID, the system looks up the descriptor using the following query:

SELECT descriptor FROM system.descriptor WHERE id = <ID>

Let's also create two new tables to use as running examples, one relatively simple, and one a little more complex. The first table is just a list of stores, with a "store_id" primary key that is an automatically incremented unique integer as the primary key (the "SERIAL" datatype) and a name.

CREATE DATABASE test;
SET DATABASE TO test;

Create TABLE stores (
  "store_id" SERIAL PRIMARY KEY,
  "name" CHAR UNIQUE
);

The second table

CREATE TABLE inventory (
  "item_id" INT UNIQUE,
  "name" CHAR UNIQUE,
  "at_store" INT,
  "stock" INT,
  PRIMARY KEY (item_id, at_store),
  CONSTRAINT at_store_fk FOREIGN KEY (at_store) REFERENCES stores (store_id)
);

Primary Key Addressing

All of the SQL data stored in tables is mapped down to individual keys and values. We call the exact mapping converting any table or row to a key value pair "key addressing". Cockroach's key addressing relies upon a primary key, and thus all tables have a primary key, whether explicitly listed in the schema or automatically generated. Note that the notion of a "primary key" refers to the primary key in the SQL sense, and is unrelated to the "key" in Cockroach's underlying key-value pairs.

Primary keys consist of one or more non-NULL columns from the table. For a given row of the table, the columns for the primary key are encoded into a single string. For example, our inventory table would be encoded as:

/item_id/at_store

[Note that "/" is being used to disambiguate the components of the key. The actual encodings do not use the "/" character. The actual encoding is specified in the `util` package in `util/encoding`. These encoding routines allow for the encoding of NULL values, integers, floating point numbers and strings such that the lexicographic ordering of the encoded strings corresponds to the same ordering of the unencoded data.]

Before being stored in the monolithic key-value space, the encoded primary key columns are prefixed with the table ID and an ID indicating that the key corresponds to the primary index. The prefix for the inventory table looks like this:

/TableID/PrimaryIndexID/item_id/at_store

Each column value is stored in a key with that prefix. Every column has a unique ID (local to the table). The value for every cell is stored at the key:

/TableID/PrimaryIndexID/item_id/at_store/ColumnID -> ColumnValue

Thus, the scan over the range

[/TableID/PrimaryIndexID/item_id/at_store,
/TableID/PrimaryIndexID/item_id/at_storf)

Where the abuse of notation "namf" in the end key refers to the key resulting from incrementing the value of the start key. As an efficiency, we do not store columns NULL values. Thus, all returned rows from the above scan give us enough information to construct the entire row. However, a row that has exclusively NULL values in non-primary key columns would have nothing stored at all. Thus, to note the existence of a row with only a primary key and remaining NULLs, every row also has a sentinel key indicating its existence. The sentinel key is simply the primary index key, with an empty value:

/TableID/PrimaryIndexID/item_id/at_store -> <empty>

Thus the above scan on such a row would return a single key, which we can use to reconstruct the row filling in NULLs for the non-primary-key values.

Column Families

The above structure is inefficient if we have many columns, since each row in an N-column table results in up to N+1 entries (1 sentinel key + N keys if every column was non-NULL). Thus, Cockroach has the ability to group multiple columns together and write them as a single key-value pair. We call this a "column family", and there are more details in this blog post: https://www.cockroachlabs.com/blog/sql-cockroachdb-column-families/

Secondary Indexes

Despite not being a formal part of the SQL standard, secondary indexes are one of its most powerful features. Secondary indexes are a level of indirection that allow quick lookups of a row using something other than the primary key. As an example, here is a secondary index on the "inventory" table, using only the "name" column:

CREATE INDEX name ON inventory (name);

This secondary index allows fast lookups based on just the "name". We use the following key addressing scheme for this non-unique index:

/TableId/SecondaryIndexID/name/item_id/at_store -> <empty>

Notice that while the index is on "name", the key contains both "name" and the values for item_id and at_store. This is done to ensure that each row for a table has a unique key for the non-unique index. In general, in order to guarantee that a non-unique index is unique, we encode the index's columns followed by any primary key columns that have not already been mentioned. Since the primary key must uniquely define a row, this transforms any non-unique index into a unique index.

Let's suppose that we had instead defined the index as:

CREATE UNIQUE INDEX name ON inventory (name, item_id);

Since this index is defined on creation as a unique index, we do not need to append the rest of the primary key columns to ensure uniqueness; instead, any insertion of a row into the table that would result in a duplication in the index will fail (and if there already are duplicates upon creation, the index creation itself will fail). However, we still need to be able to decode the full primary key by reading this index, as we will see later, in order to read any columns that are not in this index:

SELECT at_store FROM inventory WHERE name = "foo";

The solution is to put any remaining primary key columns into the value. Thus, the key addressing for this unique index looks like this:

/TableID/SecondaryIndexID/name/item_id -> at_store

The value for a unique index is composed of any primary key columns that are not already part of the index ("at_store" in this example). The goal of this key addressing scheme is to ensure that the primary key is fully specified by the key-value pair, and that the key portion is unique. However, any lookup of a non-primary and non-index column requires two reads, first to decode the primary key, and then to read the full row for the primary key, which contains all the columns. For instance, to read the value of the "stock" column in this table:

SELECT stock FROM inventory WHERE name = "foo";

Looking this up by the index on "name" does not give us the value of the "stock" column. Instead, to process this query, Cockroach does two key-value reads, which are morally equivalent to the following two SQL queries:

SELECT (item_id, at_store) FROM inventory WHERE name = "foo";

Then we use the values for the primary key that we received from the first query to perform the lookup:

SELECT stock FROM inventory WHERE item_id = "..." AND at_store = "...";

Query Planning and Execution

SQL queries are executed by converting every SQL query into a set of transactional key-value operations. The Cockroach distributed transactional key-value store provides a few operations, of which we shall discuss execution using two important ones: conditional puts, and ordered scans.

Query planning is the system which takes a parsed SQL statement (described by an abstract syntax tree) and creates an execution plan which is itself a tree of operations. The execution tree consists of leaf nodes that are SCANs and PUTs, and internal nodes that consist of operations such as join, groupby, sort, or projection.. For the bulk of SQL statements, query planning is straightforward: the complexity lies in SELECT.

At one end of the performance spectrum, an implementation of SELECT can be straightforward: do a full scan of the (joined) tables in the FROM clause, filter rows based on the WHERE clause, group the resulting rows based on the GROUP BY clause, filter those rows using the HAVING clause, and sort the remaining rows using the ORDER BY clause. There are a number of steps, but they all have well defined semantics and are mostly just an exercise in software engineering: retrieve the rows as quickly as possible and then send them through the pipeline of filtering, grouping, filtering and sorting.

However, this naive execution plan would have poor performance if the first scans return large amounts of data: if we are scanning orders of magnitude extra data, only to discard the vast majority of rows as we filter out the few rows that we need, this is needlessly inefficient. Instead, the query planner attempts to take advantage of secondary indexes to limit the data retrieved by the leafs. Additionally, the query planner makes joins between tables faster by taking advantage of the different sort orders of various secondary indexes, and avoiding re-sorting (or taking advantage of partial sorts to limit the amount of sorting done). As query planning is under active development, the details of how we implement this are in flux and will continue to be in flux for the foreseeable future. This section is intended to provide a high-level overview of a few of the techniques involved.

For a SELECT query, after parsing it, the query planner performs semantic analysis to statically verify if the query obeys basic type-safety checks, and to resolve names within the query to actual objects within the system. Let's consider a query which looks up the stock of an item in the inventory table named "foo" with item_id X:

SELECT stock FROM inventory WHERE item_id = X AND name = 'test'

The query planner first needs to resolve the "inventory" qualified name in the FROM clause to the appropriate TableDescriptor. It also needs to resolve the "item_id", "stock" and "name" column references to the appropriate column descriptions with the "inventory" TableDescriptor. Lastly, as part of semantic analysis, the query planner verifies that the expressions in the select targets and the WHERE clause are valid (e.g. the WHERE clause evaluates to a boolean).

From that starting point, the query planner then analyzes the GROUP BY and ORDER BY clauses, adding "hidden" targets for expressions used in those clauses that are not explicit targets of the query. Our example query does not have any GROUP BY or ORDER BY clauses, so we move straight to the next step: index selection. Index selection is the stage where the query planner selects the best index to scan and selects the start and end keys that minimize the amount of scanned data. Depending on the complexity of the query, the query planner might even select multiple ranges to scan from an index or multiple ranges from different indexes.

How does the query planner decide which index to use and which range of the index to scan? We currently use a restricted form of value propagation in order to determine the range of possible values for columns referenced in the WHERE clause. Using this range information, each index is examined to determine if it is a potential candidate and ranked according to its specificity. In addition to ranking indexes by the column value range information, they are ranked by how well they match the sorting required by the ORDER BY clause. A more detailed description is here: https://www.cockroachlabs.com/blog/index-selection-cockroachdb-2/, but back to the example above, the range information would determine that:

item_id >= 0 AND item_id <= 0 AND name >= 'test' and name <= 'test

Since there are two indexes on the "inventory" table, one index on "name" and another unique index on "item_id" and "name", the latter is selected as the candidate for performing a scan. To perform this scan, we need a start (inclusive) and end key (exclusive). The start key is computed using the SecondaryIndexID of the chosen index, and the constraints on the range information above:

/inventory/SecondaryIndexID/item_id/name

The end key is:

/inventory/SecondaryIndexID/item_id/namf

The "namf" suffix is not a typo: it is an abuse of notation to demonstrate how we calculate the end key: the end key is computed by incrementing the final byte of the start key such that "t" becomes "u".

Our example scan will return two key-value pairs:

/system.descriptor/primary/0/test    -> NULL
/system.descriptor/primary/0/test/id -> <ID>

The first key is the sentinel key, and the value from the second key returned by the scan is the result we need to return as the result of this SQL query.

Index

Package Files

advancecode_string.go alter_index.go alter_sequence.go alter_table.go alter_user.go analyze_expr.go app_stats.go apply_join.go authorization.go backfill.go buffer.go cancel_queries.go cancel_sessions.go check.go cluster_wide_id.go comment_on_column.go comment_on_database.go comment_on_index.go comment_on_table.go conn_executor.go conn_executor_exec.go conn_executor_prepare.go conn_fsm.go conn_io.go control_jobs.go copy.go crdb_internal.go create_database.go create_index.go create_sequence.go create_stats.go create_table.go create_user.go create_view.go data_source.go database.go deallocate.go delayed.go delete.go delete_range.go descriptor.go discard.go distinct.go distsql_physical_planner.go distsql_plan_backfill.go distsql_plan_csv.go distsql_plan_ctas.go distsql_plan_join.go distsql_plan_scrub_physical.go distsql_plan_stats.go distsql_plan_window.go distsql_running.go doc.go drop_database.go drop_index.go drop_sequence.go drop_table.go drop_user.go drop_view.go error_if_rows.go event_log.go exec_log.go exec_util.go execute.go executor_statement_metrics.go explain_distsql.go explain_plan.go explain_tree.go explain_vec.go export.go expr_filter.go filter.go grant_revoke.go group.go index_join.go information_schema.go insert.go internal.go join.go join_predicate.go lease.go limit.go logical_schema_accessors.go lookup_join.go max_one_row.go mem_metrics.go opaque.go opt_catalog.go opt_exec_factory.go opt_index_selection.go ordinality.go partition_utils.go pg_catalog.go physical_schema_accessors.go plan.go plan_batch.go plan_columns.go plan_node_to_row_source.go plan_opt.go plan_ordering.go planhook.go planner.go prepared_stmt.go project_set.go recursive_cte.go relocate.go rename_column.go rename_database.go rename_index.go rename_table.go render.go resolver.go revert.go row_source_to_plan_node.go save_table.go scan.go scatter.go schema_accessors.go schema_changer.go scrub.go scrub_constraint.go scrub_fk.go scrub_index.go scrub_physical.go select_name_resolution.go sequence.go sequence_select.go serial.go set_cluster_setting.go set_default_isolation.go set_session_authorization.go set_transaction.go set_var.go set_zone_config.go show_cluster_setting.go show_create.go show_create_clauses.go show_fingerprints.go show_histogram.go show_ranges.go show_stats.go show_trace.go show_trace_replica.go show_zone_config.go sort.go split.go spool.go statement.go subquery.go table.go tablewriter.go tablewriter_delete.go tablewriter_insert.go tablewriter_update.go tablewriter_upsert_opt.go testutils.go transform_srf.go truncate.go txn_state.go txnevent_string.go txntype_string.go unary.go union.go unsplit.go unsupported_vars.go update.go upsert.go user.go values.go vars.go views.go virtual_schema.go virtual_table.go walk.go window.go zero.go zigzag_join.go zone_config.go

Constants

const (
    NoTxnStr              = "NoTxn"
    OpenStateStr          = "Open"
    AbortedStateStr       = "Aborted"
    CommitWaitStateStr    = "CommitWait"
    RestartWaitStateStr   = "RestartWait"
    InternalErrorStateStr = "InternalError"
)

Constants for the String() representation of the session states. Shared with the CLI code which needs to recognize them.

const (
    // RunningStatusDrainingNames is for jobs that are currently in progress and
    // are draining names.
    RunningStatusDrainingNames jobs.RunningStatus = "draining names"
    // RunningStatusWaitingGC is for jobs that are currently in progress and
    // are waiting for the GC interval to expire
    RunningStatusWaitingGC jobs.RunningStatus = "waiting for GC TTL"
    // RunningStatusCompaction is for jobs that are currently in progress and
    // undergoing RocksDB compaction
    RunningStatusCompaction jobs.RunningStatus = "RocksDB compaction"
    // RunningStatusDeleteOnly is for jobs that are currently waiting on
    // the cluster to converge to seeing the schema element in the DELETE_ONLY
    // state.
    RunningStatusDeleteOnly jobs.RunningStatus = "waiting in DELETE-ONLY"
    // RunningStatusDeleteAndWriteOnly is for jobs that are currently waiting on
    // the cluster to converge to seeing the schema element in the
    // DELETE_AND_WRITE_ONLY state.
    RunningStatusDeleteAndWriteOnly jobs.RunningStatus = "waiting in DELETE-AND-WRITE_ONLY"
    // RunningStatusBackfill is for jobs that are currently running a backfill
    // for a schema element.
    RunningStatusBackfill jobs.RunningStatus = "populating schema"
    // RunningStatusValidation is for jobs that are currently validating
    // a schema element.
    RunningStatusValidation jobs.RunningStatus = "validating schema"
)
const (

    // OmitFKClausesFromCreate will not include any foreign key information in the
    // create statement.
    OmitFKClausesFromCreate shouldOmitFKClausesFromCreate
    // IncludeFkClausesInCreate will include foreign key information in the create
    // statement, and error if a FK cannot be resolved.
    IncludeFkClausesInCreate
    // OmitMissingFKClausesFromCreate will include foreign key information only if they
    // can be resolved. If not, it will ignore those constraints.
    // This is used in the case when showing the create statement for
    // tables stored in backups. Not all relevant tables may have been
    // included in the back up, so some foreign key information may be
    // impossible to retrieve.
    OmitMissingFKClausesFromCreate
)
const (
    // PgServerVersion is the latest version of postgres that we claim to support.
    PgServerVersion = "9.5.0"
    // PgServerVersionNum is the latest version of postgres that we claim to support in the numeric format of "server_version_num".
    PgServerVersionNum = "90500"
)
const FailedHashedValue = "unknown"

FailedHashedValue is used as a default return value for when HashForReporting cannot hash a value correctly.

const MaxSQLBytes = 1000

MaxSQLBytes is the maximum length in bytes of SQL statements serialized into a serverpb.Session. Exported for testing.

const ReorderJoinsLimitClusterSettingName = "sql.defaults.reorder_joins_limit"

ReorderJoinsLimitClusterSettingName is the name of the cluster setting for the maximum number of joins to reorder.

const RestartSavepointName string = "cockroach_restart"

RestartSavepointName is the only savepoint ident that we accept.

const RevertTableDefaultBatchSize = 500000

RevertTableDefaultBatchSize is the default batch size for reverting tables. This only needs to be small enough to keep raft/rocks happy -- there is no reply size to worry about. TODO(dt): tune this via experimentation.

const TableTruncateChunkSize = indexTruncateChunkSize

TableTruncateChunkSize is the maximum number of keys deleted per chunk during a table truncation.

Variables

var (
    // ErrLimitedResultNotSupported is an error produced by pgwire
    // indicating an unsupported feature of row count limits was attempted.
    ErrLimitedResultNotSupported = unimplemented.NewWithIssue(40195, "multiple active portals not supported")
    // ErrLimitedResultClosed is a sentinel error produced by pgwire
    // indicating the portal should be closed without error.
    ErrLimitedResultClosed = errors.New("row count limit closed")
)
var (
    MetaSQLExecLatency = metric.Metadata{
        Name:        "sql.exec.latency",
        Help:        "Latency of SQL statement execution",
        Measurement: "Latency",
        Unit:        metric.Unit_NANOSECONDS,
    }
    MetaSQLServiceLatency = metric.Metadata{
        Name:        "sql.service.latency",
        Help:        "Latency of SQL request execution",
        Measurement: "Latency",
        Unit:        metric.Unit_NANOSECONDS,
    }
    MetaSQLOpt = metric.Metadata{
        Name:        "sql.optimizer.count",
        Help:        "Number of statements which ran with the cost-based optimizer",
        Measurement: "SQL Statements",
        Unit:        metric.Unit_COUNT,
    }
    MetaSQLOptFallback = metric.Metadata{
        Name:        "sql.optimizer.fallback.count",
        Help:        "Number of statements which the cost-based optimizer was unable to plan",
        Measurement: "SQL Statements",
        Unit:        metric.Unit_COUNT,
    }
    MetaSQLOptPlanCacheHits = metric.Metadata{
        Name:        "sql.optimizer.plan_cache.hits",
        Help:        "Number of non-prepared statements for which a cached plan was used",
        Measurement: "SQL Statements",
        Unit:        metric.Unit_COUNT,
    }
    MetaSQLOptPlanCacheMisses = metric.Metadata{
        Name:        "sql.optimizer.plan_cache.misses",
        Help:        "Number of non-prepared statements for which a cached plan was not used",
        Measurement: "SQL Statements",
        Unit:        metric.Unit_COUNT,
    }
    MetaDistSQLSelect = metric.Metadata{
        Name:        "sql.distsql.select.count",
        Help:        "Number of DistSQL SELECT statements",
        Measurement: "SQL Statements",
        Unit:        metric.Unit_COUNT,
    }
    MetaDistSQLExecLatency = metric.Metadata{
        Name:        "sql.distsql.exec.latency",
        Help:        "Latency of DistSQL statement execution",
        Measurement: "Latency",
        Unit:        metric.Unit_NANOSECONDS,
    }
    MetaDistSQLServiceLatency = metric.Metadata{
        Name:        "sql.distsql.service.latency",
        Help:        "Latency of DistSQL request execution",
        Measurement: "Latency",
        Unit:        metric.Unit_NANOSECONDS,
    }
    MetaTxnAbort = metric.Metadata{
        Name:        "sql.txn.abort.count",
        Help:        "Number of SQL transaction abort errors",
        Measurement: "SQL Statements",
        Unit:        metric.Unit_COUNT,
    }
    MetaFailure = metric.Metadata{
        Name:        "sql.failure.count",
        Help:        "Number of statements resulting in a planning or runtime error",
        Measurement: "SQL Statements",
        Unit:        metric.Unit_COUNT,
    }
    MetaSQLTxnLatency = metric.Metadata{
        Name:        "sql.txn.latency",
        Help:        "Latency of SQL transactions",
        Measurement: "Latency",
        Unit:        metric.Unit_NANOSECONDS,
    }

    MetaQueryStarted = metric.Metadata{
        Name:        "sql.query.started.count",
        Help:        "Number of SQL queries started",
        Measurement: "SQL Statements",
        Unit:        metric.Unit_COUNT,
    }
    MetaTxnBeginStarted = metric.Metadata{
        Name:        "sql.txn.begin.started.count",
        Help:        "Number of SQL transaction BEGIN statements started",
        Measurement: "SQL Statements",
        Unit:        metric.Unit_COUNT,
    }
    MetaTxnCommitStarted = metric.Metadata{
        Name:        "sql.txn.commit.started.count",
        Help:        "Number of SQL transaction COMMIT statements started",
        Measurement: "SQL Statements",
        Unit:        metric.Unit_COUNT,
    }
    MetaTxnRollbackStarted = metric.Metadata{
        Name:        "sql.txn.rollback.started.count",
        Help:        "Number of SQL transaction ROLLBACK statements started",
        Measurement: "SQL Statements",
        Unit:        metric.Unit_COUNT,
    }
    MetaSelectStarted = metric.Metadata{
        Name:        "sql.select.started.count",
        Help:        "Number of SQL SELECT statements started",
        Measurement: "SQL Statements",
        Unit:        metric.Unit_COUNT,
    }
    MetaUpdateStarted = metric.Metadata{
        Name:        "sql.update.started.count",
        Help:        "Number of SQL UPDATE statements started",
        Measurement: "SQL Statements",
        Unit:        metric.Unit_COUNT,
    }
    MetaInsertStarted = metric.Metadata{
        Name:        "sql.insert.started.count",
        Help:        "Number of SQL INSERT statements started",
        Measurement: "SQL Statements",
        Unit:        metric.Unit_COUNT,
    }
    MetaDeleteStarted = metric.Metadata{
        Name:        "sql.delete.started.count",
        Help:        "Number of SQL DELETE statements started",
        Measurement: "SQL Statements",
        Unit:        metric.Unit_COUNT,
    }
    MetaSavepointStarted = metric.Metadata{
        Name:        "sql.savepoint.started.count",
        Help:        "Number of SQL SAVEPOINT statements started",
        Measurement: "SQL Statements",
        Unit:        metric.Unit_COUNT,
    }
    MetaRestartSavepointStarted = metric.Metadata{
        Name:        "sql.restart_savepoint.started.count",
        Help:        "Number of `SAVEPOINT cockroach_restart` statements started",
        Measurement: "SQL Statements",
        Unit:        metric.Unit_COUNT,
    }
    MetaReleaseRestartSavepointStarted = metric.Metadata{
        Name:        "sql.restart_savepoint.release.started.count",
        Help:        "Number of `RELEASE SAVEPOINT cockroach_restart` statements started",
        Measurement: "SQL Statements",
        Unit:        metric.Unit_COUNT,
    }
    MetaRollbackToRestartSavepointStarted = metric.Metadata{
        Name:        "sql.restart_savepoint.rollback.started.count",
        Help:        "Number of `ROLLBACK TO SAVEPOINT cockroach_restart` statements started",
        Measurement: "SQL Statements",
        Unit:        metric.Unit_COUNT,
    }
    MetaDdlStarted = metric.Metadata{
        Name:        "sql.ddl.started.count",
        Help:        "Number of SQL DDL statements started",
        Measurement: "SQL Statements",
        Unit:        metric.Unit_COUNT,
    }
    MetaMiscStarted = metric.Metadata{
        Name:        "sql.misc.started.count",
        Help:        "Number of other SQL statements started",
        Measurement: "SQL Statements",
        Unit:        metric.Unit_COUNT,
    }

    // Below are the metadata for the statement executed counters.
    MetaQueryExecuted = metric.Metadata{
        Name:        "sql.query.count",
        Help:        "Number of SQL queries executed",
        Measurement: "SQL Statements",
        Unit:        metric.Unit_COUNT,
    }
    MetaTxnBeginExecuted = metric.Metadata{
        Name:        "sql.txn.begin.count",
        Help:        "Number of SQL transaction BEGIN statements successfully executed",
        Measurement: "SQL Statements",
        Unit:        metric.Unit_COUNT,
    }
    MetaTxnCommitExecuted = metric.Metadata{
        Name:        "sql.txn.commit.count",
        Help:        "Number of SQL transaction COMMIT statements successfully executed",
        Measurement: "SQL Statements",
        Unit:        metric.Unit_COUNT,
    }
    MetaTxnRollbackExecuted = metric.Metadata{
        Name:        "sql.txn.rollback.count",
        Help:        "Number of SQL transaction ROLLBACK statements successfully executed",
        Measurement: "SQL Statements",
        Unit:        metric.Unit_COUNT,
    }
    MetaSelectExecuted = metric.Metadata{
        Name:        "sql.select.count",
        Help:        "Number of SQL SELECT statements successfully executed",
        Measurement: "SQL Statements",
        Unit:        metric.Unit_COUNT,
    }
    MetaUpdateExecuted = metric.Metadata{
        Name:        "sql.update.count",
        Help:        "Number of SQL UPDATE statements successfully executed",
        Measurement: "SQL Statements",
        Unit:        metric.Unit_COUNT,
    }
    MetaInsertExecuted = metric.Metadata{
        Name:        "sql.insert.count",
        Help:        "Number of SQL INSERT statements successfully executed",
        Measurement: "SQL Statements",
        Unit:        metric.Unit_COUNT,
    }
    MetaDeleteExecuted = metric.Metadata{
        Name:        "sql.delete.count",
        Help:        "Number of SQL DELETE statements successfully executed",
        Measurement: "SQL Statements",
        Unit:        metric.Unit_COUNT,
    }
    MetaSavepointExecuted = metric.Metadata{
        Name:        "sql.savepoint.count",
        Help:        "Number of SQL SAVEPOINT statements successfully executed",
        Measurement: "SQL Statements",
        Unit:        metric.Unit_COUNT,
    }
    MetaRestartSavepointExecuted = metric.Metadata{
        Name:        "sql.restart_savepoint.count",
        Help:        "Number of `SAVEPOINT cockroach_restart` statements successfully executed",
        Measurement: "SQL Statements",
        Unit:        metric.Unit_COUNT,
    }
    MetaReleaseRestartSavepointExecuted = metric.Metadata{
        Name:        "sql.restart_savepoint.release.count",
        Help:        "Number of `RELEASE SAVEPOINT cockroach_restart` statements successfully executed",
        Measurement: "SQL Statements",
        Unit:        metric.Unit_COUNT,
    }
    MetaRollbackToRestartSavepointExecuted = metric.Metadata{
        Name:        "sql.restart_savepoint.rollback.count",
        Help:        "Number of `ROLLBACK TO SAVEPOINT cockroach_restart` statements successfully executed",
        Measurement: "SQL Statements",
        Unit:        metric.Unit_COUNT,
    }
    MetaDdlExecuted = metric.Metadata{
        Name:        "sql.ddl.count",
        Help:        "Number of SQL DDL statements successfully executed",
        Measurement: "SQL Statements",
        Unit:        metric.Unit_COUNT,
    }
    MetaMiscExecuted = metric.Metadata{
        Name:        "sql.misc.count",
        Help:        "Number of other SQL statements successfully executed",
        Measurement: "SQL Statements",
        Unit:        metric.Unit_COUNT,
    }
)

Fully-qualified names for metrics.

var BoundTxnStateTransitions = fsm.Compile(fsm.Pattern{
    stateOpen{ImplicitTxn: fsm.False, RetryIntent: fsm.False}: {

        eventNonRetriableErr{IsCommit: fsm.Any}: {
            Next: stateInternalError{},
            Action: func(args fsm.Args) error {
                ts := args.Extended.(*txnState)
                ts.finishSQLTxn()
                ts.setAdvanceInfo(skipBatch, noRewind, txnAborted)
                return nil
            },
        },
        eventRetriableErr{CanAutoRetry: fsm.Any, IsCommit: fsm.False}: {
            Next: stateInternalError{},
            Action: func(args fsm.Args) error {
                ts := args.Extended.(*txnState)
                ts.finishSQLTxn()
                ts.setAdvanceInfo(skipBatch, noRewind, txnAborted)
                return nil
            },
        },
    },
})

BoundTxnStateTransitions is the state machine used by the InternalExecutor when running SQL inside a higher-level txn. It's a very limited state machine: it doesn't allow starting or finishing txns, auto-retries, etc.

var ClusterOrganization = settings.RegisterStringSetting(
    "cluster.organization",
    "organization name",
    "",
)

ClusterOrganization is the organization name.

var ClusterSecret = func() *settings.StringSetting {
    s := settings.RegisterStringSetting(
        "cluster.secret",
        "cluster specific secret",
        "",
    )
    s.SetConfidential()
    return s
}()

ClusterSecret is a cluster specific secret. This setting is hidden.

var CreatePartitioningCCL = func(
    ctx context.Context,
    st *cluster.Settings,
    evalCtx *tree.EvalContext,
    tableDesc *sqlbase.MutableTableDescriptor,
    indexDesc *sqlbase.IndexDescriptor,
    partBy *tree.PartitionBy,
) (sqlbase.PartitioningDescriptor, error) {
    return sqlbase.PartitioningDescriptor{}, sqlbase.NewCCLRequiredError(errors.New(
        "creating or manipulating partitions requires a CCL binary"))
}

CreatePartitioningCCL is the public hook point for the CCL-licensed partitioning creation code.

var DistSQLClusterExecMode = settings.RegisterEnumSetting(
    "sql.defaults.distsql",
    "default distributed SQL execution mode",
    "auto",
    map[int64]string{
        int64(sessiondata.DistSQLOff):  "off",
        int64(sessiondata.DistSQLAuto): "auto",
        int64(sessiondata.DistSQLOn):   "on",
    },
)

DistSQLClusterExecMode controls the cluster default for when DistSQL is used.

var ReorderJoinsLimitClusterValue = settings.RegisterValidatedIntSetting(
    ReorderJoinsLimitClusterSettingName,
    "default number of joins to reorder",
    opt.DefaultJoinOrderLimit,
    func(v int64) error {
        if v < 0 {
            return pgerror.Newf(pgcode.InvalidParameterValue,
                "cannot set sql.defaults.reorder_joins_limit to a negative value: %d", v)
        }
        return nil
    },
)

ReorderJoinsLimitClusterValue controls the cluster default for the maximum number of joins reordered.

var ReplicaOraclePolicy = replicaoracle.BinPackingChoice

ReplicaOraclePolicy controls which policy the physical planner uses to choose a replica for a given range. It is exported so that it may be overwritten during initialization by CCL code to enable follower reads.

var SerialNormalizationMode = settings.RegisterEnumSetting(
    "sql.defaults.serial_normalization",
    "default handling of SERIAL in table definitions",
    "rowid",
    map[int64]string{
        int64(sessiondata.SerialUsesRowID):            "rowid",
        int64(sessiondata.SerialUsesVirtualSequences): "virtual_sequence",
        int64(sessiondata.SerialUsesSQLSequences):     "sql_sequence",
    },
)

SerialNormalizationMode controls how the SERIAL type is interpreted in table definitions.

var TxnStateTransitions = fsm.Compile(fsm.Pattern{

    stateNoTxn{}: {
        eventTxnStart{fsm.Var("implicitTxn")}: {
            Description: "BEGIN, or before a statement running as an implicit txn",
            Next:        stateOpen{ImplicitTxn: fsm.Var("implicitTxn"), RetryIntent: fsm.False},
            Action: func(args fsm.Args) error {
                return args.Extended.(*txnState).noTxnToOpen(
                    args.Ctx, args.Event.(eventTxnStart),
                    args.Payload.(eventTxnStartPayload))
            },
        },
        eventNonRetriableErr{IsCommit: fsm.Any}: {

            Description: "anything but BEGIN or extended protocol command error",
            Next:        stateNoTxn{},
            Action: func(args fsm.Args) error {
                ts := args.Extended.(*txnState)
                ts.setAdvanceInfo(skipBatch, noRewind, noEvent)
                return nil
            },
        },
    },

    stateOpen{ImplicitTxn: fsm.Any, RetryIntent: fsm.Any}: {
        eventTxnFinish{}: {
            Description: "COMMIT/ROLLBACK, or after a statement running as an implicit txn",
            Next:        stateNoTxn{},
            Action: func(args fsm.Args) error {
                return args.Extended.(*txnState).finishTxn(
                    args.Payload.(eventTxnFinishPayload),
                )
            },
        },

        eventRetriableErr{CanAutoRetry: fsm.False, IsCommit: fsm.True}: {
            Description: "Retriable err on COMMIT",
            Next:        stateNoTxn{},
            Action:      cleanupAndFinish,
        },
        eventNonRetriableErr{IsCommit: fsm.True}: {
            Next:   stateNoTxn{},
            Action: cleanupAndFinish,
        },
    },
    stateOpen{ImplicitTxn: fsm.Var("implicitTxn"), RetryIntent: fsm.Var("retryIntent")}: {

        eventRetriableErr{CanAutoRetry: fsm.True, IsCommit: fsm.Any}: {

            Description: "Retriable err; will auto-retry",
            Next:        stateOpen{ImplicitTxn: fsm.Var("implicitTxn"), RetryIntent: fsm.Var("retryIntent")},
            Action: func(args fsm.Args) error {

                args.Extended.(*txnState).setAdvanceInfo(
                    rewind,
                    args.Payload.(eventRetriableErrPayload).rewCap,
                    txnRestart)
                return nil
            },
        },
    },

    stateOpen{ImplicitTxn: fsm.True, RetryIntent: fsm.False}: {
        eventRetriableErr{CanAutoRetry: fsm.False, IsCommit: fsm.False}: {
            Next:   stateNoTxn{},
            Action: cleanupAndFinish,
        },
        eventNonRetriableErr{IsCommit: fsm.False}: {
            Next:   stateNoTxn{},
            Action: cleanupAndFinish,
        },
    },

    stateOpen{ImplicitTxn: fsm.False, RetryIntent: fsm.Var("retryIntent")}: {
        eventNonRetriableErr{IsCommit: fsm.False}: {
            Next: stateAborted{RetryIntent: fsm.Var("retryIntent")},
            Action: func(args fsm.Args) error {
                ts := args.Extended.(*txnState)
                ts.mu.txn.CleanupOnError(ts.Ctx, args.Payload.(payloadWithError).errorCause())
                ts.setAdvanceInfo(skipBatch, noRewind, txnAborted)
                ts.txnAbortCount.Inc(1)
                return nil
            },
        },

        eventRetryIntentSet{}: {
            Description: "SAVEPOINT cockroach_restart",
            Next:        stateOpen{ImplicitTxn: fsm.False, RetryIntent: fsm.True},
            Action: func(args fsm.Args) error {

                args.Extended.(*txnState).setAdvanceInfo(advanceOne, noRewind, noEvent)
                return nil
            },
        },
    },
    stateOpen{ImplicitTxn: fsm.False, RetryIntent: fsm.False}: {

        eventRetriableErr{CanAutoRetry: fsm.False, IsCommit: fsm.False}: {
            Description: "RetryIntent not set, so handled like non-retriable err",
            Next:        stateAborted{RetryIntent: fsm.False},
            Action: func(args fsm.Args) error {
                ts := args.Extended.(*txnState)
                ts.mu.txn.CleanupOnError(ts.Ctx, args.Payload.(payloadWithError).errorCause())
                ts.setAdvanceInfo(skipBatch, noRewind, txnAborted)
                ts.txnAbortCount.Inc(1)
                return nil
            },
        },
    },
    stateOpen{ImplicitTxn: fsm.False, RetryIntent: fsm.True}: {
        eventRetriableErr{CanAutoRetry: fsm.False, IsCommit: fsm.False}: {
            Next: stateRestartWait{},
            Action: func(args fsm.Args) error {

                args.Extended.(*txnState).setAdvanceInfo(skipBatch, noRewind, txnRestart)
                return nil
            },
        },
        eventTxnReleased{}: {
            Description: "RELEASE SAVEPOINT cockroach_restart",
            Next:        stateCommitWait{},
            Action: func(args fsm.Args) error {
                args.Extended.(*txnState).setAdvanceInfo(advanceOne, noRewind, txnCommit)
                return nil
            },
        },

        eventTxnRestart{}: {
            Description: "ROLLBACK TO SAVEPOINT cockroach_restart",
            Next:        stateOpen{ImplicitTxn: fsm.False, RetryIntent: fsm.True},
            Action: func(args fsm.Args) error {
                state := args.Extended.(*txnState)

                state.mu.txn.ManualRestart(args.Ctx, hlc.Timestamp{})
                args.Extended.(*txnState).setAdvanceInfo(advanceOne, noRewind, txnRestart)
                return nil
            },
        },
    },

    stateAborted{RetryIntent: fsm.Var("retryIntent")}: {
        eventTxnFinish{}: {
            Description: "ROLLBACK",
            Next:        stateNoTxn{},
            Action: func(args fsm.Args) error {
                return args.Extended.(*txnState).finishTxn(
                    args.Payload.(eventTxnFinishPayload),
                )
            },
        },
        eventNonRetriableErr{IsCommit: fsm.Any}: {

            Description: "any other statement",
            Next:        stateAborted{RetryIntent: fsm.Var("retryIntent")},
            Action: func(args fsm.Args) error {
                args.Extended.(*txnState).setAdvanceInfo(skipBatch, noRewind, noEvent)
                return nil
            },
        },
    },
    stateAborted{RetryIntent: fsm.True}: {

        eventTxnStart{ImplicitTxn: fsm.False}: {
            Description: "ROLLBACK TO SAVEPOINT cockroach_restart",
            Next:        stateOpen{ImplicitTxn: fsm.False, RetryIntent: fsm.True},
            Action: func(args fsm.Args) error {
                ts := args.Extended.(*txnState)
                ts.finishSQLTxn()

                payload := args.Payload.(eventTxnStartPayload)

                ts.resetForNewSQLTxn(
                    ts.connCtx,
                    explicitTxn,
                    payload.txnSQLTimestamp,
                    payload.historicalTimestamp,
                    payload.pri, payload.readOnly,
                    nil,
                    args.Payload.(eventTxnStartPayload).tranCtx,
                )
                ts.setAdvanceInfo(advanceOne, noRewind, noEvent)
                return nil
            },
        },
    },

    stateRestartWait{}: {

        eventTxnFinish{}: {
            Description: "ROLLBACK",
            Next:        stateNoTxn{},
            Action: func(args fsm.Args) error {
                return args.Extended.(*txnState).finishTxn(
                    args.Payload.(eventTxnFinishPayload),
                )
            },
        },

        eventTxnRestart{}: {
            Description: "ROLLBACK TO SAVEPOINT cockroach_restart",
            Next:        stateOpen{ImplicitTxn: fsm.False, RetryIntent: fsm.True},
            Action: func(args fsm.Args) error {
                args.Extended.(*txnState).setAdvanceInfo(advanceOne, noRewind, txnRestart)
                return nil
            },
        },
        eventNonRetriableErr{IsCommit: fsm.Any}: {
            Next: stateAborted{RetryIntent: fsm.True},
            Action: func(args fsm.Args) error {
                ts := args.Extended.(*txnState)
                ts.mu.txn.CleanupOnError(ts.Ctx, args.Payload.(eventNonRetriableErrPayload).err)
                ts.setAdvanceInfo(skipBatch, noRewind, txnAborted)
                ts.txnAbortCount.Inc(1)
                return nil
            },
        },
    },

    stateCommitWait{}: {
        eventTxnFinish{}: {
            Description: "COMMIT",
            Next:        stateNoTxn{},
            Action: func(args fsm.Args) error {
                return args.Extended.(*txnState).finishTxn(
                    args.Payload.(eventTxnFinishPayload),
                )
            },
        },
        eventNonRetriableErr{IsCommit: fsm.Any}: {

            Description: "any other statement",
            Next:        stateCommitWait{},
            Action: func(args fsm.Args) error {
                args.Extended.(*txnState).setAdvanceInfo(skipBatch, noRewind, noEvent)
                return nil
            },
        },
    },
})

TxnStateTransitions describe the transitions used by a connExecutor's fsm.Machine. Args.Extended is a txnState, which is muted by the Actions.

This state machine accepts the eventNonRetriableErr{IsCommit: fsm.True} in all states. This contract is in place to support the cleanup of connExecutor -> this event can always be sent when the connExecutor is tearing down.

NOTE: The Args.Ctx passed to the actions is the connExecutor's context. While we are inside a SQL txn, the txn's ctx should be used for operations (i.e txnState.Ctx, which is a child ctx). This is so because transitions that move in and out of transactions need to have access to both contexts.

go:generate ../util/fsm/gen/reports.sh TxnStateTransitions stateNoTxn

var UnsupportedVars = func(ss ...string) map[string]struct{} {
    m := map[string]struct{}{}
    for _, s := range ss {
        m[s] = struct{}{}
    }
    return m
}(

    "optimize_bounded_sort",

    "array_nulls",
    "backend_flush_after",
    "backslash_quote",

    "check_function_bodies",

    "commit_delay",
    "commit_siblings",
    "constraint_exclusion",
    "cpu_index_tuple_cost",
    "cpu_operator_cost",
    "cpu_tuple_cost",
    "cursor_tuple_fraction",
    "deadlock_timeout",
    "debug_deadlocks",
    "debug_pretty_print",
    "debug_print_parse",
    "debug_print_plan",
    "debug_print_rewritten",
    "default_statistics_target",
    "default_text_search_config",
    "default_transaction_deferrable",

    "default_with_oids",
    "dynamic_library_path",
    "effective_cache_size",
    "enable_bitmapscan",
    "enable_gathermerge",
    "enable_hashagg",
    "enable_hashjoin",
    "enable_indexonlyscan",
    "enable_indexscan",
    "enable_material",
    "enable_mergejoin",
    "enable_nestloop",
    "enable_seqscan",
    "enable_sort",
    "enable_tidscan",
    "escape_string_warning",
    "exit_on_error",

    "force_parallel_mode",
    "from_collapse_limit",
    "geqo",
    "geqo_effort",
    "geqo_generations",
    "geqo_pool_size",
    "geqo_seed",
    "geqo_selection_bias",
    "geqo_threshold",
    "gin_fuzzy_search_limit",
    "gin_pending_list_limit",

    "ignore_checksum_failure",
    "join_collapse_limit",
    "lc_messages",
    "lc_monetary",
    "lc_numeric",
    "lc_time",
    "lo_compat_privileges",
    "local_preload_libraries",

    "log_btree_build_stats",
    "log_duration",
    "log_error_verbosity",
    "log_executor_stats",
    "log_lock_waits",
    "log_min_duration_statement",
    "log_min_error_statement",
    "log_min_messages",
    "log_parser_stats",
    "log_planner_stats",
    "log_replication_commands",
    "log_statement",
    "log_statement_stats",
    "log_temp_files",
    "maintenance_work_mem",
    "max_parallel_workers",
    "max_parallel_workers_per_gather",
    "max_stack_depth",
    "min_parallel_index_scan_size",
    "min_parallel_table_scan_size",
    "operator_precedence_warning",
    "parallel_setup_cost",
    "parallel_tuple_cost",
    "password_encryption",
    "quote_all_identifiers",
    "random_page_cost",
    "replacement_sort_tuples",
    "role",

    "seed",
    "seq_page_cost",

    "session_preload_libraries",
    "session_replication_role",

    "synchronous_commit",
    "tcp_keepalives_count",
    "tcp_keepalives_idle",
    "tcp_keepalives_interval",
    "temp_buffers",
    "temp_file_limit",
    "temp_tablespaces",
    "timezone_abbreviations",
    "trace_lock_oidmin",
    "trace_lock_table",
    "trace_locks",
    "trace_lwlocks",
    "trace_notify",
    "trace_sort",
    "trace_syncscan",
    "trace_userlocks",
    "track_activities",
    "track_counts",
    "track_functions",
    "track_io_timing",
    "transaction_deferrable",

    "transform_null_equals",
    "update_process_title",
    "vacuum_cost_delay",
    "vacuum_cost_limit",
    "vacuum_cost_page_dirty",
    "vacuum_cost_page_hit",
    "vacuum_cost_page_miss",
    "vacuum_freeze_min_age",
    "vacuum_freeze_table_age",
    "vacuum_multixact_freeze_min_age",
    "vacuum_multixact_freeze_table_age",
    "wal_compression",
    "wal_consistency_checking",
    "wal_debug",
    "work_mem",
    "xmlbinary",
    "xmloption",
    "zero_damaged_pages",
)

UnsupportedVars contains the set of PostgreSQL session variables and client parameters that are not supported in CockroachDB. These are used to produce error messages and telemetry.

var VectorizeClusterMode = settings.RegisterEnumSetting(
    "sql.defaults.vectorize",
    "default vectorize mode",
    "auto",
    map[int64]string{
        int64(sessiondata.VectorizeOff):            "off",
        int64(sessiondata.VectorizeAuto):           "auto",
        int64(sessiondata.VectorizeExperimentalOn): "experimental_on",
    },
)

VectorizeClusterMode controls the cluster default for when automatic vectorization is enabled.

var VectorizeRowCountThresholdClusterValue = settings.RegisterValidatedIntSetting(
    "sql.defaults.vectorize_row_count_threshold",
    "default vectorize row count threshold",
    colexec.DefaultVectorizeRowCountThreshold,
    func(v int64) error {
        if v < 0 {
            return pgerror.Newf(pgcode.InvalidParameterValue,
                "cannot set sql.defaults.vectorize_row_count_threshold to a negative value: %d", v)
        }
        return nil
    },
)

VectorizeRowCountThresholdClusterValue controls the cluster default for the vectorize row count threshold. When it is met, the vectorized execution engine will be used if possible.

func AddPlanHook Uses

func AddPlanHook(f planHookFn)

AddPlanHook adds a hook used to short-circuit creating a planNode from a tree.Statement. If the func returned by the hook is non-nil, it is used to construct a planNode that runs that func in a goroutine during Start.

See PlanHookState comments for information about why plan hooks are needed.

func AddWrappedPlanHook Uses

func AddWrappedPlanHook(f wrappedPlanHookFn)

AddWrappedPlanHook adds a hook used to short-circuit creating a planNode from a tree.Statement. If the returned plan is non-nil, it is used directly by the planner.

See PlanHookState comments for information about why plan hooks are needed.

func AnonymizeStatementsForReporting Uses

func AnonymizeStatementsForReporting(action, sqlStmts string, r interface{}) error

AnonymizeStatementsForReporting transforms an action, SQL statements, and a value (usually a recovered panic) into an error that will be useful when passed to our error reporting as it exposes a scrubbed version of the statements.

func CheckPrivilegeForUser Uses

func CheckPrivilegeForUser(
    _ context.Context, user string, descriptor sqlbase.DescriptorProto, privilege privilege.Kind,
) error

CheckPrivilegeForUser verifies that `user“ has `privilege` on `descriptor`. This is not part of the planner as the only caller (ccl/sqlccl/restore.go) does not have one.

func CountLeases Uses

func CountLeases(
    ctx context.Context, executor sqlutil.InternalExecutor, tables []IDVersion, at hlc.Timestamp,
) (int, error)

CountLeases returns the number of unexpired leases for a number of tables each at a particular version at a particular time.

func CreatePartitioning Uses

func CreatePartitioning(
    ctx context.Context,
    st *cluster.Settings,
    evalCtx *tree.EvalContext,
    tableDesc *sqlbase.MutableTableDescriptor,
    indexDesc *sqlbase.IndexDescriptor,
    partBy *tree.PartitionBy,
) (sqlbase.PartitioningDescriptor, error)

CreatePartitioning constructs the partitioning descriptor for an index that is partitioned into ranges, each addressable by zone configs.

func CreateTestTableDescriptor Uses

func CreateTestTableDescriptor(
    ctx context.Context,
    parentID, id sqlbase.ID,
    schema string,
    privileges *sqlbase.PrivilegeDescriptor,
) (sqlbase.TableDescriptor, error)

CreateTestTableDescriptor converts a SQL string to a table for test purposes. Will fail on complex tables where that operation requires e.g. looking up other tables.

func DistIngest Uses

func DistIngest(
    ctx context.Context,
    phs PlanHookState,
    job *jobs.Job,
    tables map[string]*execinfrapb.ReadImportDataSpec_ImportTable,
    from []string,
    format roachpb.IOFileFormat,
    walltime int64,
) (roachpb.BulkOpSummary, error)

DistIngest is used by IMPORT to run a DistSQL flow to ingest data by starting reader processes on many nodes that each read and ingest their assigned files and then send back a summary of what they ingested. The combined summary is returned.

func GenerateSubzoneSpans Uses

func GenerateSubzoneSpans(
    st *cluster.Settings,
    clusterID uuid.UUID,
    tableDesc *sqlbase.TableDescriptor,
    subzones []config.Subzone,
    hasNewSubzones bool,
) ([]config.SubzoneSpan, error)

GenerateSubzoneSpans constructs from a TableDescriptor the entries mapping zone config spans to subzones for use in the SubzoneSpans field of config.ZoneConfig. SubzoneSpans controls which splits are created, so only the spans corresponding to entries in subzones are returned.

Zone configs target indexes and partitions via `subzones`, which are attached to a table-scoped row in `system.zones`. Each subzone represents one index (primary or secondary) or one partition (or subpartition) and contains the usual zone config constraints. They are saved to `system.zones` sparsely (only when set by a user) and are the most specific entry in the normal cluster-default/database/table/subzone config hierarchy.

Each non-interleaved index and partition can be mapped to spans in the keyspace. Indexes and range partitions each map to one span, while each list partition maps to one or more spans. Each partition span is contained by some index span and each subpartition span is contained by one of its parent partition's spans. The spans for a given level of a range partitioning (corresponding to one `PARTITION BY` in sql or one `PartitionDescriptor`) are disjoint, but the spans for a given level of a list partitioning may overlap if DEFAULT is used. A list partitioning which includes both (1, DEFAULT) and (1, 2) will overlap with the latter getting precedence in the zone config hierarchy. NB: In a valid PartitionDescriptor, no partitions with the same number of DEFAULTs will overlap (this property is used by `indexCoveringsForPartitioning`).

These subzone spans are kept denormalized to the relevant `system.zone` row for performance. Given a TableDescriptor, the spans for every index/partition/subpartition are created, filtered out if they don't have a config set for them, and precedence applied (via `OverlapCoveringMerge`) to produce a set of non-overlapping spans, which each map to a subzone. There may be "holes" (uncovered spans) in this set.

The returned spans are returned in exactly the format required by `system.zones`. They must be sorted and non-overlapping. Each contains an IndexID, which maps to one of the input `subzones` by indexing into the slice. As space optimizations, all `Key`s and `EndKey`s of `SubzoneSpan` omit the common prefix (the encoded table ID) and if `EndKey` is equal to `Key.PrefixEnd()` it is omitted.

This function has tests in the partitionccl package.

TODO(benesch): remove the hasNewSubzones parameter when a statement to clear all subzones at once is introduced.

func GenerateUniqueDescID Uses

func GenerateUniqueDescID(ctx context.Context, db *client.DB) (sqlbase.ID, error)

GenerateUniqueDescID returns the next available Descriptor ID and increments the counter. The incrementing is non-transactional, and the counter could be incremented multiple times because of retries.

func GetAllDatabaseDescriptorIDs Uses

func GetAllDatabaseDescriptorIDs(ctx context.Context, txn *client.Txn) ([]sqlbase.ID, error)

GetAllDatabaseDescriptorIDs looks up and returns all available database descriptor IDs.

func GetAllDescriptors Uses

func GetAllDescriptors(ctx context.Context, txn *client.Txn) ([]sqlbase.DescriptorProto, error)

GetAllDescriptors looks up and returns all available descriptors.

func GetUserHashedPassword Uses

func GetUserHashedPassword(
    ctx context.Context, ie *InternalExecutor, metrics *MemoryMetrics, username string,
) (bool, []byte, error)

GetUserHashedPassword returns the hashedPassword for the given username if found in system.users.

func GetZoneConfigInTxn Uses

func GetZoneConfigInTxn(
    ctx context.Context,
    txn *client.Txn,
    id uint32,
    index *sqlbase.IndexDescriptor,
    partition string,
    getInheritedDefault bool,
) (uint32, *config.ZoneConfig, *config.Subzone, error)

GetZoneConfigInTxn looks up the zone and subzone for the specified object ID, index, and partition.

func HashForReporting Uses

func HashForReporting(secret, appName string) string

HashForReporting 1-way hashes values for use in stat reporting. The secret should be the cluster.secret setting.

func InitTableDescriptor Uses

func InitTableDescriptor(
    id, parentID sqlbase.ID,
    name string,
    creationTime hlc.Timestamp,
    privileges *sqlbase.PrivilegeDescriptor,
    temporary bool,
) sqlbase.MutableTableDescriptor

InitTableDescriptor returns a blank TableDescriptor.

func IsSessionVariableConfigurable Uses

func IsSessionVariableConfigurable(varName string) (exists, configurable bool)

IsSessionVariableConfigurable returns true iff there is a session variable with the given name and it is settable by a client (e.g. in pgwire).

func MakeCheckConstraint Uses

func MakeCheckConstraint(
    ctx context.Context,
    desc *sqlbase.MutableTableDescriptor,
    d *tree.CheckConstraintTableDef,
    inuseNames map[string]struct{},
    semaCtx *tree.SemaContext,
    tableName tree.TableName,
) (*sqlbase.TableDescriptor_CheckConstraint, error)

MakeCheckConstraint makes a descriptor representation of a check from a def.

func MakeIndexDescriptor Uses

func MakeIndexDescriptor(n *tree.CreateIndex) (*sqlbase.IndexDescriptor, error)

MakeIndexDescriptor creates an index descriptor from a CreateIndex node.

func MakeSequenceKeyVal Uses

func MakeSequenceKeyVal(
    sequence *TableDescriptor, newVal int64, isCalled bool,
) ([]byte, int64, error)

MakeSequenceKeyVal returns the key and value of a sequence being set with newVal.

func MakeSequenceTableDesc Uses

func MakeSequenceTableDesc(
    sequenceName string,
    sequenceOptions tree.SequenceOptions,
    parentID sqlbase.ID,
    id sqlbase.ID,
    creationTime hlc.Timestamp,
    privileges *sqlbase.PrivilegeDescriptor,
    settings *cluster.Settings,
) (sqlbase.MutableTableDescriptor, error)

MakeSequenceTableDesc creates a sequence descriptor.

func MakeTableDesc Uses

func MakeTableDesc(
    ctx context.Context,
    txn *client.Txn,
    vt SchemaResolver,
    st *cluster.Settings,
    n *tree.CreateTable,
    parentID, id sqlbase.ID,
    creationTime hlc.Timestamp,
    privileges *sqlbase.PrivilegeDescriptor,
    affected map[sqlbase.ID]*sqlbase.MutableTableDescriptor,
    semaCtx *tree.SemaContext,
    evalCtx *tree.EvalContext,
    temporary bool,
) (sqlbase.MutableTableDescriptor, error)

MakeTableDesc creates a table descriptor from a CreateTable statement.

txn and vt can be nil if the table to be created does not contain references to other tables (e.g. foreign keys or interleaving). This is useful at bootstrap when creating descriptors for virtual tables.

evalCtx can be nil if the table to be created has no default expression for any of the columns and no partitioning expression.

semaCtx can be nil if the table to be created has no default expression on any of the columns and no check constraints.

The caller must also ensure that the SchemaResolver is configured to bypass caching and enable visibility of just-added descriptors. This is used to resolve sequence and FK dependencies. Also see the comment at the start of the global scope resolveFK().

If the table definition *may* use the SERIAL type, the caller is also responsible for processing serial types using processSerialInColumnDef() on every column definition, and creating the necessary sequences in KV before calling MakeTableDesc().

func MustGetDatabaseDescByID Uses

func MustGetDatabaseDescByID(
    ctx context.Context, txn *client.Txn, id sqlbase.ID,
) (*sqlbase.DatabaseDescriptor, error)

MustGetDatabaseDescByID looks up the database descriptor given its ID, returning an error if the descriptor is not found.

func NewInternalPlanner Uses

func NewInternalPlanner(
    opName string, txn *client.Txn, user string, memMetrics *MemoryMetrics, execCfg *ExecutorConfig,
) (interface{}, func())

NewInternalPlanner is an exported version of newInternalPlanner. It returns an interface{} so it can be used outside of the sql package.

func NormalizeAndValidateUsername Uses

func NormalizeAndValidateUsername(username string) (string, error)

NormalizeAndValidateUsername case folds the specified username and verifies it validates according to the usernameRE regular expression. It rejects reserved user names.

func NormalizeAndValidateUsernameNoBlacklist Uses

func NormalizeAndValidateUsernameNoBlacklist(username string) (string, error)

NormalizeAndValidateUsernameNoBlacklist case folds the specified username and verifies it validates according to the usernameRE regular expression.

func ParseHLC Uses

func ParseHLC(s string) (hlc.Timestamp, error)

ParseHLC parses a string representation of an `hlc.Timestamp`.

func PlanAndRunCTAS Uses

func PlanAndRunCTAS(
    ctx context.Context,
    dsp *DistSQLPlanner,
    planner *planner,
    txn *client.Txn,
    isLocal bool,
    in planNode,
    out execinfrapb.ProcessorCoreUnion,
    recv *DistSQLReceiver,
)

PlanAndRunCTAS plans and runs the CREATE TABLE AS command.

func ResolveFK Uses

func ResolveFK(
    ctx context.Context,
    txn *client.Txn,
    sc SchemaResolver,
    tbl *sqlbase.MutableTableDescriptor,
    d *tree.ForeignKeyConstraintTableDef,
    backrefs map[sqlbase.ID]*sqlbase.MutableTableDescriptor,
    ts FKTableState,
    validationBehavior tree.ValidationBehavior,
    settings *cluster.Settings,
) error

ResolveFK looks up the tables and columns mentioned in a `REFERENCES` constraint and adds metadata representing that constraint to the descriptor. It may, in doing so, add to or alter descriptors in the passed in `backrefs` map of other tables that need to be updated when this table is created. Constraints that are not known to hold for existing data are created "unvalidated", but when table is empty (e.g. during creation), no existing data implies no existing violations, and thus the constraint can be created without the unvalidated flag.

The caller should pass an instance of fkSelfResolver as SchemaResolver, so that FK references can find the newly created table for self-references.

The caller must also ensure that the SchemaResolver is configured to bypass caching and enable visibility of just-added descriptors. If there are any FKs, the descriptor of the depended-on table must be looked up uncached, and we'll allow FK dependencies on tables that were just added.

The passed Txn is used to lookup databases to qualify names in error messages but if nil, will result in unqualified names in those errors.

The passed validationBehavior is used to determine whether or not preexisting entries in the table need to be validated against the foreign key being added. This only applies for existing tables, not new tables.

func RevertTables Uses

func RevertTables(
    ctx context.Context,
    db *client.DB,
    tables []*sqlbase.TableDescriptor,
    targetTime hlc.Timestamp,
    batchSize int64,
) error

RevertTables reverts the passed table to the target time.

func ScanMetaKVs Uses

func ScanMetaKVs(
    ctx context.Context, txn *client.Txn, span roachpb.Span,
) ([]client.KeyValue, error)

ScanMetaKVs returns the meta KVs for the ranges that touch the given span.

func ShowCreate Uses

func ShowCreate(
    ctx context.Context,
    dbPrefix string,
    allDescs []sqlbase.Descriptor,
    desc *sqlbase.TableDescriptor,
    ignoreFKs shouldOmitFKClausesFromCreate,
) (string, error)

ShowCreate returns a valid SQL representation of the CREATE statement used to create the descriptor passed in. The

The names of the tables references by foreign keys, and the interleaved parent if any, are prefixed by their own database name unless it is equal to the given dbPrefix. This allows us to elide the prefix when the given table references other tables in the current database.

func ShowCreatePartitioning Uses

func ShowCreatePartitioning(
    a *sqlbase.DatumAlloc,
    tableDesc *sqlbase.TableDescriptor,
    idxDesc *sqlbase.IndexDescriptor,
    partDesc *sqlbase.PartitioningDescriptor,
    buf *bytes.Buffer,
    indent int,
    colOffset int,
) error

ShowCreatePartitioning returns a PARTITION BY clause for the specified index, if applicable.

func ShowCreateSequence Uses

func ShowCreateSequence(
    ctx context.Context, tn *tree.Name, desc *sqlbase.TableDescriptor,
) (string, error)

ShowCreateSequence returns a valid SQL representation of the CREATE SEQUENCE statement used to create the given sequence.

func ShowCreateTable Uses

func ShowCreateTable(
    ctx context.Context,
    tn *tree.Name,
    dbPrefix string,
    desc *sqlbase.TableDescriptor,
    lCtx *internalLookupCtx,
    fkDisplayMode shouldOmitFKClausesFromCreate,
) (string, error)

ShowCreateTable returns a valid SQL representation of the CREATE TABLE statement used to create the given table.

The names of the tables references by foreign keys, and the interleaved parent if any, are prefixed by their own database name unless it is equal to the given dbPrefix. This allows us to elide the prefix when the given table references other tables in the current database.

func ShowCreateView Uses

func ShowCreateView(
    ctx context.Context, tn *tree.Name, desc *sqlbase.TableDescriptor,
) (string, error)

ShowCreateView returns a valid SQL representation of the CREATE VIEW statement used to create the given view. It is used in the implementation of the crdb_internal.create_statements virtual table.

func SimplifySerialInColumnDefWithRowID Uses

func SimplifySerialInColumnDefWithRowID(
    ctx context.Context, d *tree.ColumnTableDef, tableName *ObjectName,
) error

SimplifySerialInColumnDefWithRowID analyzes a column definition and simplifies any use of SERIAL as if SerialNormalizationMode was set to SerialUsesRowID. No sequence needs to be created.

This is currently used by bulk I/O import statements which do not (yet?) support customization of the SERIAL behavior.

func TestDisableTableLeases Uses

func TestDisableTableLeases() func()

TestDisableTableLeases disables table leases and returns a function that can be used to enable it.

func TestingOverrideExplainEnvVersion Uses

func TestingOverrideExplainEnvVersion(ver string) func()

TestingOverrideExplainEnvVersion overrides the version reported by EXPLAIN (OPT, ENV). Used for testing.

func WriteNewDescToBatch Uses

func WriteNewDescToBatch(
    ctx context.Context,
    kvTrace bool,
    s *cluster.Settings,
    b *client.Batch,
    tableID sqlbase.ID,
    desc sqlbase.DescriptorProto,
) (err error)

WriteNewDescToBatch adds a CPut command writing a descriptor proto to the descriptors table. It writes the descriptor desc at the id descID, asserting that there was no previous descriptor at that id present already. If kvTrace is enabled, it will log an event explaining the CPut that was performed.

func ZoneConfigHook Uses

func ZoneConfigHook(
    cfg *config.SystemConfig, id uint32,
) (*config.ZoneConfig, *config.ZoneConfig, bool, error)

ZoneConfigHook returns the zone config for the object with id using the cached system config. If keySuffix is within a subzone, the subzone's config is returned instead. The bool is set to true when the value returned is cached.

type AuthorizationAccessor Uses

type AuthorizationAccessor interface {
    // CheckPrivilege verifies that the user has `privilege` on `descriptor`.
    CheckPrivilege(
        ctx context.Context, descriptor sqlbase.DescriptorProto, privilege privilege.Kind,
    ) error

    // CheckAnyPrivilege returns nil if user has any privileges at all.
    CheckAnyPrivilege(ctx context.Context, descriptor sqlbase.DescriptorProto) error

    // HasAdminRole returns tuple of bool and error:
    // (true, nil) means that the user has an admin role (i.e. root or node)
    // (false, nil) means that the user has NO admin role
    // (false, err) means that there was an error running the query on
    // the `system.users` table
    HasAdminRole(ctx context.Context) (bool, error)

    // RequireAdminRole is a wrapper on top of HasAdminRole.
    // It errors if HasAdminRole errors or if the user isn't a super-user.
    // Includes the named action in the error message.
    RequireAdminRole(ctx context.Context, action string) error

    // MemberOfWithAdminOption looks up all the roles (direct and indirect) that 'member' is a member
    // of and returns a map of role -> isAdmin.
    MemberOfWithAdminOption(ctx context.Context, member string) (map[string]bool, error)
}

AuthorizationAccessor for checking authorization (e.g. desc privileges).

type BindResult Uses

type BindResult interface {
    ResultBase
}

BindResult represents the result of a Bind command.

type BindStmt Uses

type BindStmt struct {
    PreparedStatementName string
    PortalName            string
    // OutFormats contains the requested formats for the output columns.
    // It either contains a bunch of format codes, in which case the number will
    // need to match the number of output columns of the portal, or contains a single
    // code, in which case that code will be applied to all columns.
    OutFormats []pgwirebase.FormatCode
    // Args are the arguments for the prepared statement.
    // They are passed in without decoding because decoding requires type
    // inference to have been performed.
    //
    // A nil element means a tree.DNull argument.
    Args [][]byte
    // ArgFormatCodes are the codes to be used to deserialize the Args.
    // It either contains a bunch of format codes, in which case the number will
    // need to match the number of arguments for the portal, or contains a single
    // code, in which case that code will be applied to all arguments.
    ArgFormatCodes []pgwirebase.FormatCode
    // contains filtered or unexported fields
}

BindStmt is the Command for creating a portal from a prepared statement.

func (BindStmt) String Uses

func (b BindStmt) String() string

type CachedPhysicalAccessor Uses

type CachedPhysicalAccessor struct {
    SchemaAccessor
    // contains filtered or unexported fields
}

CachedPhysicalAccessor adds a cache on top of any SchemaAccessor.

func (*CachedPhysicalAccessor) GetDatabaseDesc Uses

func (a *CachedPhysicalAccessor) GetDatabaseDesc(
    ctx context.Context, txn *client.Txn, name string, flags tree.DatabaseLookupFlags,
) (desc *DatabaseDescriptor, err error)

GetDatabaseDesc implements the SchemaAccessor interface.

func (*CachedPhysicalAccessor) GetObjectDesc Uses

func (a *CachedPhysicalAccessor) GetObjectDesc(
    ctx context.Context, txn *client.Txn, name *ObjectName, flags tree.ObjectLookupFlags,
) (ObjectDescriptor, error)

GetObjectDesc implements the SchemaAccessor interface.

type ClientComm Uses

type ClientComm interface {
    // createStatementResult creates a StatementResult for stmt.
    //
    // descOpt specifies if result needs to inform the client about row schema. If
    // it doesn't, a SetColumns call becomes a no-op.
    //
    // pos is the stmt's position within the connection and is used to enforce
    // that results are created in order and also to discard results through
    // ClientLock.rtrim(pos).
    //
    // formatCodes describe how each column in the result rows is to be encoded.
    // It should be nil if statement type != Rows. Otherwise, it can be nil, in
    // which case every column will be encoded using the text encoding, otherwise
    // it needs to contain a value for every column.
    CreateStatementResult(
        stmt tree.Statement,
        descOpt RowDescOpt,
        pos CmdPos,
        formatCodes []pgwirebase.FormatCode,
        conv sessiondata.DataConversionConfig,
        limit int,
        portalName string,
        implicitTxn bool,
    ) CommandResult
    // CreatePrepareResult creates a result for a PrepareStmt command.
    CreatePrepareResult(pos CmdPos) ParseResult
    // CreateDescribeResult creates a result for a DescribeStmt command.
    CreateDescribeResult(pos CmdPos) DescribeResult
    // CreateBindResult creates a result for a BindStmt command.
    CreateBindResult(pos CmdPos) BindResult
    // CreateDeleteResult creates a result for a DeletePreparedStmt command.
    CreateDeleteResult(pos CmdPos) DeleteResult
    // CreateSyncResult creates a result for a Sync command.
    CreateSyncResult(pos CmdPos) SyncResult
    // CreateFlushResult creates a result for a Flush command.
    CreateFlushResult(pos CmdPos) FlushResult
    // CreateErrorResult creates a result on which only errors can be communicated
    // to the client.
    CreateErrorResult(pos CmdPos) ErrorResult
    // CreateEmptyQueryResult creates a result for an empty-string query.
    CreateEmptyQueryResult(pos CmdPos) EmptyQueryResult
    // CreateCopyInResult creates a result for a Copy-in command.
    CreateCopyInResult(pos CmdPos) CopyInResult
    // CreateDrainResult creates a result for a Drain command.
    CreateDrainResult(pos CmdPos) DrainResult

    // lockCommunication ensures that no further results are delivered to the
    // client. The returned ClientLock can be queried to see what results have
    // been already delivered to the client and to discard results that haven't
    // been delivered.
    //
    // ClientLock.Close() needs to be called on the returned lock once
    // communication can be unlocked (i.e. results can be delivered to the client
    // again).
    LockCommunication() ClientLock

    // Flush delivers all the previous results to the client. The results might
    // have been buffered, in which case this flushes the buffer.
    Flush(pos CmdPos) error
}

ClientComm is the interface used by the connExecutor for creating results to be communicated to client and for exerting some control over this communication.

ClientComm is implemented by the pgwire connection.

type ClientLock Uses

type ClientLock interface {
    // Close unlocks the ClientComm from whence this ClientLock came from. After
    // Close is called, buffered results may again be sent to the client,
    // according to the result streaming policy.
    //
    // Once Close() is called, the ClientLock cannot be used anymore.
    Close()

    // ClientPos returns the position of the latest command for which results
    // have been sent to the client. The position is relative to the start of the
    // connection.
    ClientPos() CmdPos

    // RTrim iterates backwards through the results and drops all results with
    // position >= pos.
    // It is illegal to call rtrim with a position <= clientPos(). In other words,
    // results can
    RTrim(ctx context.Context, pos CmdPos)
}

ClientLock is an interface returned by ClientComm.lockCommunication(). It represents a lock on the delivery of results to a SQL client. While such a lock is used, no more results are delivered. The lock itself can be used to query what results have already been delivered and to discard results that haven't been delivered.

type ClusterWideID Uses

type ClusterWideID struct {
    uint128.Uint128
}

ClusterWideID represents an identifier that is guaranteed to be unique across a cluster. It is a wrapper around a uint128. It logically consists of 96 bits of HLC timestamp, and 32 bits of node ID.

func BytesToClusterWideID Uses

func BytesToClusterWideID(b []byte) ClusterWideID

BytesToClusterWideID converts raw bytes into a ClusterWideID. The caller is responsible for ensuring the byte slice contains 16 bytes.

func GenerateClusterWideID Uses

func GenerateClusterWideID(timestamp hlc.Timestamp, nodeID roachpb.NodeID) ClusterWideID

GenerateClusterWideID takes a timestamp and node ID, and generates a ClusterWideID.

func StringToClusterWideID Uses

func StringToClusterWideID(s string) (ClusterWideID, error)

StringToClusterWideID converts a string to a ClusterWideID. If the string is not a valid uint128, an error is returned.

func (ClusterWideID) GetNodeID Uses

func (id ClusterWideID) GetNodeID() int32

GetNodeID extracts the node ID from a ClusterWideID.

type CmdPos Uses

type CmdPos int64

CmdPos represents the index of a command relative to the start of a connection. The first command received on a connection has position 0.

type Command Uses

type Command interface {
    fmt.Stringer
    // contains filtered or unexported methods
}

Command is an interface implemented by all commands pushed by pgwire into the buffer.

type CommandResult Uses

type CommandResult interface {
    RestrictedCommandResult
    CommandResultClose
}

CommandResult represents the result of a statement. It which needs to be ultimately delivered to the client. pgwire.conn implements this.

type CommandResultClose Uses

type CommandResultClose interface {
    // Close marks a result as complete. No further uses of the CommandResult are
    // allowed after this call. All results must be eventually closed through
    // Close()/CloseWithErr()/Discard(), except in case query processing has
    // encountered an irrecoverable error and the client connection will be
    // closed; in such cases it is not mandated that these functions are called on
    // the result that may have been open at the time the error occurred.
    // NOTE(andrei): We might want to tighten the contract if the results get any
    // state that needs to be closed even when the whole connection is about to be
    // terminated.
    Close(TransactionStatusIndicator)

    // CloseWithErr is like Close, except it tells the client that an execution
    // error has happened. All rows previously accumulated on the result might be
    // discarded; only this error might be delivered to the client as a result of
    // the command.
    //
    // After calling CloseWithErr it is illegal to create CommandResults for any
    // command in the same batch as the one being closed. The contract is that the
    // next result created corresponds to the first command in the next batch.
    CloseWithErr(err error)

    // Discard is called to mark the fact that the result is being disposed off.
    // No completion message will be sent to the client. The expectation is that
    // either the no other methods on the result had previously been used (and so
    // no data has been buffered for the client), or there is a communication lock
    // in effect and the buffer will be rewound - in either case, the client will
    // never see any bytes pertaining to this result.
    Discard()
}

CommandResultClose is a subset of CommandResult dealing with the closing of the result.

type CommandResultErrBase Uses

type CommandResultErrBase interface {
    // SetError accumulates an execution error that needs to be reported to the
    // client. No further calls other than SetError(), Close()/CloseWithError()
    // and Discard() are allowed.
    //
    // Calling SetError() a second time overwrites the previously set error.
    SetError(error)

    // Err returns the error previously set with SetError(), if any.
    Err() error
}

CommandResultErrBase is the subset of CommandResult dealing with setting a query execution error.

type ConnectionHandler Uses

type ConnectionHandler struct {
    // contains filtered or unexported fields
}

ConnectionHandler is the interface between the result of SetupConn and the ServeConn below. It encapsulates the connExecutor and hides it away from other packages.

func (ConnectionHandler) GetStatusParam Uses

func (h ConnectionHandler) GetStatusParam(ctx context.Context, varName string) string

GetStatusParam retrieves the configured value of the session variable identified by varName. This is used for the initial message sent to a client during a session set-up.

func (ConnectionHandler) GetUnqualifiedIntSize Uses

func (h ConnectionHandler) GetUnqualifiedIntSize() *types.T

GetUnqualifiedIntSize implements pgwire.sessionDataProvider and returns the type that INT should be parsed as.

func (ConnectionHandler) RegisterOnSessionDataChange Uses

func (h ConnectionHandler) RegisterOnSessionDataChange(key string, f func(val string))

RegisterOnSessionDataChange adds a listener to execute when a change on the given key is made using the mutator object.

type CopyIn Uses

type CopyIn struct {
    Stmt *tree.CopyFrom
    // Conn is the network connection. Execution of the CopyFrom statement takes
    // control of the connection.
    Conn pgwirebase.Conn
    // CopyDone is decremented once execution finishes, signaling that control of
    // the connection is being handed back to the network routine.
    CopyDone *sync.WaitGroup
}

CopyIn is the command for execution of the Copy-in pgwire subprotocol.

func (CopyIn) String Uses

func (CopyIn) String() string

type CopyInResult Uses

type CopyInResult interface {
    ResultBase
}

CopyInResult represents the result of a CopyIn command. Closing this result produces no output for the client.

type CreateUserNode Uses

type CreateUserNode struct {
    // contains filtered or unexported fields
}

CreateUserNode creates entries in the system.users table. This is called from CREATE USER and CREATE ROLE.

func (*CreateUserNode) Close Uses

func (*CreateUserNode) Close(context.Context)

Close implements the planNode interface.

func (*CreateUserNode) FastPathResults Uses

func (n *CreateUserNode) FastPathResults() (int, bool)

FastPathResults implements the planNodeFastPath interface.

func (*CreateUserNode) Next Uses

func (*CreateUserNode) Next(runParams) (bool, error)

Next implements the planNode interface.

func (*CreateUserNode) Values Uses

func (*CreateUserNode) Values() tree.Datums

Values implements the planNode interface.

type DatabaseDescriptor Uses

type DatabaseDescriptor = sqlbase.DatabaseDescriptor

DatabaseDescriptor is provided for convenience and to make the interface definitions below more intuitive.

func ResolveTargetObject Uses

func ResolveTargetObject(
    ctx context.Context, sc SchemaResolver, tn *ObjectName,
) (res *DatabaseDescriptor, err error)

ResolveTargetObject determines a valid target path for an object that may not exist yet. It returns the descriptor for the database where the target object lives.

The object name is modified in-place with the result of the name resolution.

type DeletePreparedStmt Uses

type DeletePreparedStmt struct {
    Name string
    Type pgwirebase.PrepareType
}

DeletePreparedStmt is the Command for freeing a prepared statement.

func (DeletePreparedStmt) String Uses

func (d DeletePreparedStmt) String() string

type DeleteResult Uses

type DeleteResult interface {
    ResultBase
}

DeleteResult represents the result of a DeletePreparedStatement command.

type DescribeResult Uses

type DescribeResult interface {
    ResultBase

    // SetInferredTypes tells the client about the inferred placeholder types.
    SetInferredTypes([]oid.Oid)
    // SetNoDataDescription is used to tell the client that the prepared statement
    // or portal produces no rows.
    SetNoDataRowDescription()
    // SetPrepStmtOutput tells the client about the results schema of a prepared
    // statement.
    SetPrepStmtOutput(context.Context, sqlbase.ResultColumns)
    // SetPortalOutput tells the client about the results schema and formatting of
    // a portal.
    SetPortalOutput(context.Context, sqlbase.ResultColumns, []pgwirebase.FormatCode)
}

DescribeResult represents the result of a Describe command (for either describing a prepared statement or a portal).

type DescribeStmt Uses

type DescribeStmt struct {
    Name string
    Type pgwirebase.PrepareType
}

DescribeStmt is the Command for producing info about a prepared statement or portal.

func (DescribeStmt) String Uses

func (d DescribeStmt) String() string

type DistSQLPlanner Uses

type DistSQLPlanner struct {
    // contains filtered or unexported fields
}

DistSQLPlanner is used to generate distributed plans from logical plans. A rough overview of the process:

- the plan is based on a planNode tree (in the future it will be based on an
  intermediate representation tree). Only a subset of the possible trees is
  supported (this can be checked via CheckSupport).

- we generate a PhysicalPlan for the planNode tree recursively. The
  PhysicalPlan consists of a network of processors and streams, with a set
  of unconnected "result routers". The PhysicalPlan also has information on
  ordering and on the mapping planNode columns to columns in the result
  streams (all result routers output streams with the same schema).

  The PhysicalPlan for a scanNode leaf consists of TableReaders, one for each node
  that has one or more ranges.

- for each an internal planNode we start with the plan of the child node(s)
  and add processing stages (connected to the result routers of the children
  node).

func NewDistSQLPlanner Uses

func NewDistSQLPlanner(
    ctx context.Context,
    planVersion execinfrapb.DistSQLVersion,
    st *cluster.Settings,
    nodeDesc roachpb.NodeDescriptor,
    rpcCtx *rpc.Context,
    distSQLSrv *distsql.ServerImpl,
    distSender *kv.DistSender,
    gossip *gossip.Gossip,
    stopper *stop.Stopper,
    liveness livenessProvider,
    nodeDialer *nodedialer.Dialer,
) *DistSQLPlanner

NewDistSQLPlanner initializes a DistSQLPlanner.

nodeDesc is the descriptor of the node on which this planner runs. It is used to favor itself and other close-by nodes when planning. An empty descriptor can be passed to aid bootstrapping, but then SetNodeDesc() needs to be called before this planner is used.

func (*DistSQLPlanner) CheckNodeHealthAndVersion Uses

func (dsp *DistSQLPlanner) CheckNodeHealthAndVersion(
    planCtx *PlanningCtx, desc *roachpb.NodeDescriptor,
) error

CheckNodeHealthAndVersion adds the node to planCtx if it is healthy and has a compatible version. An error is returned otherwise.

func (*DistSQLPlanner) Exec Uses

func (dsp *DistSQLPlanner) Exec(
    ctx context.Context, localPlanner interface{}, sql string, distribute bool,
) error

Exec is a test utility function that takes a localPlanner (of type interface{} so that external packages can call NewInternalPlanner and pass the result) and executes a sql statement through the DistSQLPlanner.

func (*DistSQLPlanner) FinalizePlan Uses

func (dsp *DistSQLPlanner) FinalizePlan(planCtx *PlanningCtx, plan *PhysicalPlan)

FinalizePlan adds a final "result" stage if necessary and populates the endpoints of the plan.

func (*DistSQLPlanner) NewPlanningCtx Uses

func (dsp *DistSQLPlanner) NewPlanningCtx(
    ctx context.Context, evalCtx *extendedEvalContext, txn *client.Txn,
) *PlanningCtx

NewPlanningCtx returns a new PlanningCtx.

func (*DistSQLPlanner) PartitionSpans Uses

func (dsp *DistSQLPlanner) PartitionSpans(
    planCtx *PlanningCtx, spans roachpb.Spans,
) ([]SpanPartition, error)

PartitionSpans finds out which nodes are owners for ranges touching the given spans, and splits the spans according to owning nodes. The result is a set of SpanPartitions (guaranteed one for each relevant node), which form a partitioning of the spans (i.e. they are non-overlapping and their union is exactly the original set of spans).

PartitionSpans does its best to not assign ranges on nodes that are known to either be unhealthy or running an incompatible version. The ranges owned by such nodes are assigned to the gateway.

func (*DistSQLPlanner) PlanAndRun Uses

func (dsp *DistSQLPlanner) PlanAndRun(
    ctx context.Context,
    evalCtx *extendedEvalContext,
    planCtx *PlanningCtx,
    txn *client.Txn,
    plan planNode,
    recv *DistSQLReceiver,
) (cleanup func())

PlanAndRun generates a physical plan from a planNode tree and executes it. It assumes that the tree is supported (see CheckSupport).

All errors encountered are reported to the DistSQLReceiver's resultWriter. Additionally, if the error is a "communication error" (an error encountered while using that resultWriter), the error is also stored in DistSQLReceiver.commErr. That can be tested to see if a client session needs to be closed.

It returns a non-nil (although it can be a noop when an error is encountered) cleanup function that must be called once the planTop AST is no longer needed and can be closed. Note that this function also cleans up the flow which is unfortunate but is caused by the sharing of memory monitors between planning and execution - cleaning up the flow wants to close the monitor, but it cannot do so because the AST needs to live longer and still uses the same monitor. That's why we end up in a situation that in order to clean up the flow, we need to close the AST first, but we can only do that after PlanAndRun returns.

func (*DistSQLPlanner) PlanAndRunPostqueries Uses

func (dsp *DistSQLPlanner) PlanAndRunPostqueries(
    ctx context.Context,
    planner *planner,
    evalCtxFactory func() *extendedEvalContext,
    postqueryPlans []postquery,
    recv *DistSQLReceiver,
    maybeDistribute bool,
) bool

PlanAndRunPostqueries returns false if an error was encountered and sets that error in the provided receiver.

func (*DistSQLPlanner) PlanAndRunSubqueries Uses

func (dsp *DistSQLPlanner) PlanAndRunSubqueries(
    ctx context.Context,
    planner *planner,
    evalCtxFactory func() *extendedEvalContext,
    subqueryPlans []subquery,
    recv *DistSQLReceiver,
    maybeDistribute bool,
) bool

PlanAndRunSubqueries returns false if an error was encountered and sets that error in the provided receiver.

func (*DistSQLPlanner) Run Uses

func (dsp *DistSQLPlanner) Run(
    planCtx *PlanningCtx,
    txn *client.Txn,
    plan *PhysicalPlan,
    recv *DistSQLReceiver,
    evalCtx *extendedEvalContext,
    finishedSetupFn func(),
) (cleanup func())

Run executes a physical plan. The plan should have been finalized using FinalizePlan.

All errors encountered are reported to the DistSQLReceiver's resultWriter. Additionally, if the error is a "communication error" (an error encountered while using that resultWriter), the error is also stored in DistSQLReceiver.commErr. That can be tested to see if a client session needs to be closed.

Args: - txn is the transaction in which the plan will run. If nil, the different processors are expected to manage their own internal transactions. - evalCtx is the evaluation context in which the plan will run. It might be mutated. - finishedSetupFn, if non-nil, is called synchronously after all the processors have successfully started up.

It returns a non-nil (although it can be a noop when an error is encountered) cleanup function that must be called in order to release the resources.

func (*DistSQLPlanner) SetNodeDesc Uses

func (dsp *DistSQLPlanner) SetNodeDesc(desc roachpb.NodeDescriptor)

SetNodeDesc sets the planner's node descriptor. The first call to SetNodeDesc leads to the construction of the SpanResolver.

func (*DistSQLPlanner) SetSpanResolver Uses

func (dsp *DistSQLPlanner) SetSpanResolver(spanResolver physicalplan.SpanResolver)

SetSpanResolver switches to a different SpanResolver. It is the caller's responsibility to make sure the DistSQLPlanner is not in use.

type DistSQLReceiver Uses

type DistSQLReceiver struct {
    // contains filtered or unexported fields
}

DistSQLReceiver is a RowReceiver that writes results to a rowResultWriter. This is where the DistSQL execution meets the SQL Session - the RowContainer comes from a client Session.

DistSQLReceiver also update the RangeDescriptorCache and the LeaseholderCache in response to DistSQL metadata about misplanned ranges.

func MakeDistSQLReceiver Uses

func MakeDistSQLReceiver(
    ctx context.Context,
    resultWriter rowResultWriter,
    stmtType tree.StatementType,
    rangeCache *kv.RangeDescriptorCache,
    leaseCache *kv.LeaseHolderCache,
    txn *client.Txn,
    updateClock func(observedTs hlc.Timestamp),
    tracing *SessionTracing,
) *DistSQLReceiver

MakeDistSQLReceiver creates a DistSQLReceiver.

ctx is the Context that the receiver will use throughout its lifetime. resultWriter is the container where the results will be stored. If only the row count is needed, this can be nil.

txn is the transaction in which the producer flow runs; it will be updated on errors. Nil if the flow overall doesn't run in a transaction.

func (*DistSQLReceiver) ProducerDone Uses

func (r *DistSQLReceiver) ProducerDone()

ProducerDone is part of the RowReceiver interface.

func (*DistSQLReceiver) Push Uses

func (r *DistSQLReceiver) Push(
    row sqlbase.EncDatumRow, meta *execinfrapb.ProducerMetadata,
) execinfra.ConsumerStatus

Push is part of the RowReceiver interface.

func (*DistSQLReceiver) Release Uses

func (r *DistSQLReceiver) Release()

Release releases this DistSQLReceiver back to the pool.

func (*DistSQLReceiver) SetError Uses

func (r *DistSQLReceiver) SetError(err error)

SetError provides a convenient way for a client to pass in an error, thus pretending that a query execution error happened. The error is passed along to the resultWriter.

func (*DistSQLReceiver) Types Uses

func (r *DistSQLReceiver) Types() []types.T

Types is part of the RowReceiver interface.

type DrainRequest Uses

type DrainRequest struct{}

DrainRequest represents a notice that the server is draining and command processing should stop soon.

DrainRequest commands don't produce results.

func (DrainRequest) String Uses

func (DrainRequest) String() string

type DrainResult Uses

type DrainResult interface {
    ResultBase
}

DrainResult represents the result of a Drain command. Closing this result produces no output for the client.

type DropUserNode Uses

type DropUserNode struct {
    // contains filtered or unexported fields
}

DropUserNode deletes entries from the system.users table. This is called from DROP USER and DROP ROLE.

func (*DropUserNode) Close Uses

func (*DropUserNode) Close(context.Context)

Close implements the planNode interface.

func (*DropUserNode) FastPathResults Uses

func (n *DropUserNode) FastPathResults() (int, bool)

FastPathResults implements the planNodeFastPath interface.

func (*DropUserNode) Next Uses

func (*DropUserNode) Next(runParams) (bool, error)

Next implements the planNode interface.

func (*DropUserNode) Values Uses

func (*DropUserNode) Values() tree.Datums

Values implements the planNode interface.

type EmptyQueryResult Uses

type EmptyQueryResult interface {
    ResultBase
}

EmptyQueryResult represents the result of an empty query (a query representing a blank string).

type EngineMetrics Uses

type EngineMetrics struct {
    // The subset of SELECTs that are processed through DistSQL.
    DistSQLSelectCount *metric.Counter
    // The subset of queries that are processed by the cost-based optimizer.
    SQLOptCount *metric.Counter
    // The subset of queries which we attempted and failed to plan with the
    // cost-based optimizer.
    SQLOptFallbackCount   *metric.Counter
    SQLOptPlanCacheHits   *metric.Counter
    SQLOptPlanCacheMisses *metric.Counter

    DistSQLExecLatency    *metric.Histogram
    SQLExecLatency        *metric.Histogram
    DistSQLServiceLatency *metric.Histogram
    SQLServiceLatency     *metric.Histogram
    SQLTxnLatency         *metric.Histogram

    // TxnAbortCount counts transactions that were aborted, either due
    // to non-retriable errors, or retriable errors when the client-side
    // retry protocol is not in use.
    TxnAbortCount *metric.Counter

    // FailureCount counts non-retriable errors in open transactions.
    FailureCount *metric.Counter
}

EngineMetrics groups a set of SQL metrics.

func (EngineMetrics) MetricStruct Uses

func (EngineMetrics) MetricStruct()

MetricStruct is part of the metric.Struct interface.

type ErrorResult Uses

type ErrorResult interface {
    ResultBase
}

ErrorResult represents the result of a SendError command.

type EventLogSetClusterSettingDetail Uses

type EventLogSetClusterSettingDetail struct {
    SettingName string
    Value       string
    User        string
}

EventLogSetClusterSettingDetail is the json details for a settings change.

type EventLogType Uses

type EventLogType string

EventLogType represents an event type that can be recorded in the event log.

const (
    // EventLogCreateDatabase is recorded when a database is created.
    EventLogCreateDatabase EventLogType = "create_database"
    // EventLogDropDatabase is recorded when a database is dropped.
    EventLogDropDatabase EventLogType = "drop_database"

    // EventLogCreateTable is recorded when a table is created.
    EventLogCreateTable EventLogType = "create_table"
    // EventLogDropTable is recorded when a table is dropped.
    EventLogDropTable EventLogType = "drop_table"
    // EventLogTruncateTable is recorded when a table is truncated.
    EventLogTruncateTable EventLogType = "truncate_table"
    // EventLogAlterTable is recorded when a table is altered.
    EventLogAlterTable EventLogType = "alter_table"
    // EventLogCommentOnColumn is recorded when a column is commented.
    EventLogCommentOnColumn EventLogType = "comment_on_column"
    // EventLogCommentOnTable is recorded when a table is commented.
    EventLogCommentOnDatabase EventLogType = "comment_on_database"
    // EventLogCommentOnTable is recorded when a table is commented.
    EventLogCommentOnTable EventLogType = "comment_on_table"
    // EventLogCommentOnIndex is recorded when a index is commented.
    EventLogCommentOnIndex EventLogType = "comment_on_index"

    // EventLogCreateIndex is recorded when an index is created.
    EventLogCreateIndex EventLogType = "create_index"
    // EventLogDropIndex is recorded when an index is dropped.
    EventLogDropIndex EventLogType = "drop_index"
    // EventLogAlterIndex is recorded when an index is altered.
    EventLogAlterIndex EventLogType = "alter_index"

    // EventLogCreateView is recorded when a view is created.
    EventLogCreateView EventLogType = "create_view"
    // EventLogDropView is recorded when a view is dropped.
    EventLogDropView EventLogType = "drop_view"

    // EventLogCreateSequence is recorded when a sequence is created.
    EventLogCreateSequence EventLogType = "create_sequence"
    // EventLogDropSequence is recorded when a sequence is dropped.
    EventLogDropSequence EventLogType = "drop_sequence"
    // EventLogAlterSequence is recorded when a sequence is altered.
    EventLogAlterSequence EventLogType = "alter_sequence"

    // EventLogReverseSchemaChange is recorded when an in-progress schema change
    // encounters a problem and is reversed.
    EventLogReverseSchemaChange EventLogType = "reverse_schema_change"
    // EventLogFinishSchemaChange is recorded when a previously initiated schema
    // change has completed.
    EventLogFinishSchemaChange EventLogType = "finish_schema_change"
    // EventLogFinishSchemaRollback is recorded when a previously
    // initiated schema change rollback has completed.
    EventLogFinishSchemaRollback EventLogType = "finish_schema_change_rollback"

    // EventLogNodeJoin is recorded when a node joins the cluster.
    EventLogNodeJoin EventLogType = "node_join"
    // EventLogNodeRestart is recorded when an existing node rejoins the cluster
    // after being offline.
    EventLogNodeRestart EventLogType = "node_restart"
    // EventLogNodeDecommissioned is recorded when a node is marked as
    // decommissioning.
    EventLogNodeDecommissioned EventLogType = "node_decommissioned"
    // EventLogNodeRecommissioned is recorded when a decommissioned node is
    // recommissioned.
    EventLogNodeRecommissioned EventLogType = "node_recommissioned"

    // EventLogSetClusterSetting is recorded when a cluster setting is changed.
    EventLogSetClusterSetting EventLogType = "set_cluster_setting"

    // EventLogSetZoneConfig is recorded when a zone config is changed.
    EventLogSetZoneConfig EventLogType = "set_zone_config"
    // EventLogRemoveZoneConfig is recorded when a zone config is removed.
    EventLogRemoveZoneConfig EventLogType = "remove_zone_config"

    // EventLogCreateStatistics is recorded when statistics are collected for a
    // table.
    EventLogCreateStatistics EventLogType = "create_statistics"
)

NOTE: When you add a new event type here. Please manually add it to pkg/ui/src/util/eventTypes.ts so that it will be recognized in the UI.

type EventLogger Uses

type EventLogger struct {
    *InternalExecutor
}

An EventLogger exposes methods used to record events to the event table.

func MakeEventLogger Uses

func MakeEventLogger(execCfg *ExecutorConfig) EventLogger

MakeEventLogger constructs a new EventLogger.

func (EventLogger) InsertEventRecord Uses

func (ev EventLogger) InsertEventRecord(
    ctx context.Context,
    txn *client.Txn,
    eventType EventLogType,
    targetID, reportingID int32,
    info interface{},
) error

InsertEventRecord inserts a single event into the event log as part of the provided transaction.

type ExecPortal Uses

type ExecPortal struct {
    Name string
    // limit is a feature of pgwire that we don't really support. We accept it and
    // don't complain as long as the statement produces fewer results than this.
    Limit int
    // TimeReceived is the time at which the exec message was received
    // from the client. Used to compute the service latency.
    TimeReceived time.Time
}

ExecPortal is the Command for executing a portal.

func (ExecPortal) String Uses

func (e ExecPortal) String() string

type ExecStmt Uses

type ExecStmt struct {
    // Information returned from parsing: AST, SQL, NumPlaceholders.
    // Note that AST can be nil, in which case executing it should produce an
    // "empty query response" message.
    parser.Statement

    // TimeReceived is the time at which the exec message was received
    // from the client. Used to compute the service latency.
    TimeReceived time.Time
    // ParseStart/ParseEnd are the timing info for parsing of the query. Used for
    // stats reporting.
    ParseStart time.Time
    ParseEnd   time.Time
}

ExecStmt is the command for running a query sent through the "simple" pgwire protocol.

func (ExecStmt) String Uses

func (e ExecStmt) String() string

type ExecutorConfig Uses

type ExecutorConfig struct {
    Settings *cluster.Settings
    NodeInfo
    DefaultZoneConfig *config.ZoneConfig
    Locality          roachpb.Locality
    AmbientCtx        log.AmbientContext
    DB                *client.DB
    Gossip            *gossip.Gossip
    DistSender        *kv.DistSender
    RPCContext        *rpc.Context
    LeaseManager      *LeaseManager
    Clock             *hlc.Clock
    DistSQLSrv        *distsql.ServerImpl
    StatusServer      serverpb.StatusServer
    MetricsRecorder   nodeStatusGenerator
    SessionRegistry   *SessionRegistry
    JobRegistry       *jobs.Registry
    VirtualSchemas    *VirtualSchemaHolder
    DistSQLPlanner    *DistSQLPlanner
    TableStatsCache   *stats.TableStatisticsCache
    StatsRefresher    *stats.Refresher
    ExecLogger        *log.SecondaryLogger
    AuditLogger       *log.SecondaryLogger
    InternalExecutor  *InternalExecutor
    QueryCache        *querycache.C

    TestingKnobs              ExecutorTestingKnobs
    PGWireTestingKnobs        *PGWireTestingKnobs
    SchemaChangerTestingKnobs *SchemaChangerTestingKnobs
    DistSQLRunTestingKnobs    *execinfra.TestingKnobs
    EvalContextTestingKnobs   tree.EvalContextTestingKnobs
    // HistogramWindowInterval is (server.Config).HistogramWindowInterval.
    HistogramWindowInterval time.Duration

    // Caches updated by DistSQL.
    RangeDescriptorCache *kv.RangeDescriptorCache
    LeaseHolderCache     *kv.LeaseHolderCache
}

An ExecutorConfig encompasses the auxiliary objects and configuration required to create an executor. All fields holding a pointer or an interface are required to create a Executor; the rest will have sane defaults set if omitted.

func (*ExecutorConfig) Organization Uses

func (ec *ExecutorConfig) Organization() string

Organization returns the value of cluster.organization.

type ExecutorTestingKnobs Uses

type ExecutorTestingKnobs struct {
    // StatementFilter can be used to trap execution of SQL statements and
    // optionally change their results. The filter function is invoked after each
    // statement has been executed.
    StatementFilter StatementFilter

    // BeforeExecute is called by the Executor before plan execution. It is useful
    // for synchronizing statement execution.
    BeforeExecute func(ctx context.Context, stmt string)

    // AfterExecute is like StatementFilter, but it runs in the same goroutine of the
    // statement.
    AfterExecute func(ctx context.Context, stmt string, err error)

    // DisableAutoCommit, if set, disables the auto-commit functionality of some
    // SQL statements. That functionality allows some statements to commit
    // directly when they're executed in an implicit SQL txn, without waiting for
    // the Executor to commit the implicit txn.
    // This has to be set in tests that need to abort such statements using a
    // StatementFilter; otherwise, the statement commits immediately after
    // execution so there'll be nothing left to abort by the time the filter runs.
    DisableAutoCommit bool

    // BeforeAutoCommit is called when the Executor is about to commit the KV
    // transaction after running a statement in an implicit transaction, allowing
    // tests to inject errors into that commit.
    // If an error is returned, that error will be considered the result of
    // txn.Commit(), and the txn.Commit() call will not actually be
    // made. If no error is returned, txn.Commit() is called normally.
    //
    // Note that this is not called if the SQL statement representing the implicit
    // transaction has committed the KV txn itself (e.g. if it used the 1-PC
    // optimization). This is only called when the Executor is the one doing the
    // committing.
    BeforeAutoCommit func(ctx context.Context, stmt string) error
}

ExecutorTestingKnobs is part of the context used to control parts of the system during testing.

func (*ExecutorTestingKnobs) ModuleTestingKnobs Uses

func (*ExecutorTestingKnobs) ModuleTestingKnobs()

ModuleTestingKnobs is part of the base.ModuleTestingKnobs interface.

type FKTableState Uses

type FKTableState int

FKTableState is the state of the referencing table resolveFK() is called on.

const (
    // NewTable represents a new table, where the FK constraint is specified in the
    // CREATE TABLE
    NewTable FKTableState = iota
    // EmptyTable represents an existing table that is empty
    EmptyTable
    // NonEmptyTable represents an existing non-empty table
    NonEmptyTable
)

type Flush Uses

type Flush struct{}

Flush is a Command asking for the results of all previous commands to be delivered to the client.

func (Flush) String Uses

func (Flush) String() string

type FlushResult Uses

type FlushResult interface {
    ResultBase
}

FlushResult represents the result of a Flush command. When this result is closed, all previously accumulated results are flushed to the client.

type IDVersion Uses

type IDVersion struct {
    // contains filtered or unexported fields
}

IDVersion represents a descriptor ID, version pair that are meant to map to a single immutable descriptor.

func NewIDVersionPrev Uses

func NewIDVersionPrev(desc *sqlbase.TableDescriptor) IDVersion

NewIDVersionPrev returns an initialized IDVersion with the previous version of the descriptor.

type ImmutableTableDescriptor Uses

type ImmutableTableDescriptor = sqlbase.ImmutableTableDescriptor

ImmutableTableDescriptor is provided for convenience and to make the interface definitions below more intuitive.

func ResolveExistingObject Uses

func ResolveExistingObject(
    ctx context.Context,
    sc SchemaResolver,
    tn *ObjectName,
    lookupFlags tree.ObjectLookupFlags,
    requiredType ResolveRequiredType,
) (res *ImmutableTableDescriptor, err error)

ResolveExistingObject looks up an existing object. If required is true, an error is returned if the object does not exist. Optionally, if a desired descriptor type is specified, that type is checked.

The object name is modified in-place with the result of the name resolution, if successful. It is not modified in case of error or if no object is found.

type InternalExecutor Uses

type InternalExecutor struct {
    // contains filtered or unexported fields
}

InternalExecutor can be used internally by code modules to execute SQL statements without needing to open a SQL connection.

InternalExecutor can execute one statement at a time. As of 03/2018, it doesn't offer a session interface for maintaining session state or for running explicit SQL transactions. However, it supports running SQL statements inside a higher-lever (KV) txn and inheriting session variables from another session (for the latter see SessionBoundInternalExecutor).

func MakeInternalExecutor Uses

func MakeInternalExecutor(
    ctx context.Context, s *Server, memMetrics MemoryMetrics, settings *cluster.Settings,
) InternalExecutor

MakeInternalExecutor creates an InternalExecutor.

func (*InternalExecutor) Exec Uses

func (ie *InternalExecutor) Exec(
    ctx context.Context, opName string, txn *client.Txn, stmt string, qargs ...interface{},
) (int, error)

Exec executes the supplied SQL statement. Statements are currently executed as the root user with the system database as current database.

If txn is not nil, the statement will be executed in the respective txn.

Returns the number of rows affected.

func (*InternalExecutor) ExecWithUser Uses

func (ie *InternalExecutor) ExecWithUser(
    ctx context.Context,
    opName string,
    txn *client.Txn,
    userName string,
    stmt string,
    qargs ...interface{},
) (int, error)

ExecWithUser is like Exec, except it changes the username to that specified.

func (*InternalExecutor) Query Uses

func (ie *InternalExecutor) Query(
    ctx context.Context, opName string, txn *client.Txn, stmt string, qargs ...interface{},
) ([]tree.Datums, error)

Query executes the supplied SQL statement and returns the resulting rows. The statement is executed as the root user.

If txn is not nil, the statement will be executed in the respective txn.

func (*InternalExecutor) QueryRow Uses

func (ie *InternalExecutor) QueryRow(
    ctx context.Context, opName string, txn *client.Txn, stmt string, qargs ...interface{},
) (tree.Datums, error)

QueryRow is like Query, except it returns a single row, or nil if not row is found, or an error if more that one row is returned.

func (*InternalExecutor) QueryWithCols Uses

func (ie *InternalExecutor) QueryWithCols(
    ctx context.Context, opName string, txn *client.Txn, stmt string, qargs ...interface{},
) ([]tree.Datums, sqlbase.ResultColumns, error)

QueryWithCols is like Query, but it also returns the computed ResultColumns of the input query.

func (*InternalExecutor) QueryWithUser Uses

func (ie *InternalExecutor) QueryWithUser(
    ctx context.Context,
    opName string,
    txn *client.Txn,
    userName string,
    stmt string,
    qargs ...interface{},
) ([]tree.Datums, sqlbase.ResultColumns, error)

QueryWithUser is like Query, except it changes the username to that specified.

type KVStringOptValidate Uses

type KVStringOptValidate string

KVStringOptValidate indicates the requested validation of a TypeAsStringOpts option.

const (
    KVStringOptAny            KVStringOptValidate = `any`
    KVStringOptRequireNoValue KVStringOptValidate = `no-value`
    KVStringOptRequireValue   KVStringOptValidate = `value`
)

KVStringOptValidate values

type LeaseAcquireBlockType Uses

type LeaseAcquireBlockType int

LeaseAcquireBlockType is the type of blocking result event when calling LeaseAcquireResultBlockEvent.

const (
    // LeaseAcquireBlock denotes the LeaseAcquireResultBlockEvent is
    // coming from tableState.acquire().
    LeaseAcquireBlock LeaseAcquireBlockType = iota
    // LeaseAcquireFreshestBlock denotes the LeaseAcquireResultBlockEvent is
    // from tableState.acquireFreshestFromStore().
    LeaseAcquireFreshestBlock
)

type LeaseManager Uses

type LeaseManager struct {
    LeaseStore
    // contains filtered or unexported fields
}

LeaseManager manages acquiring and releasing per-table leases. It also handles resolving table names to descriptor IDs. The leases are managed internally with a table descriptor and expiration time exported by the API. The table descriptor acquired needs to be released. A transaction can use a table descriptor as long as its timestamp is within the validity window for the descriptor: descriptor.ModificationTime <= txn.Timestamp < expirationTime

Exported only for testing.

The locking order is: LeaseManager.mu > tableState.mu > tableNameCache.mu > tableVersionState.mu

func NewLeaseManager Uses

func NewLeaseManager(
    ambientCtx log.AmbientContext,
    nodeIDContainer *base.NodeIDContainer,
    db *client.DB,
    clock *hlc.Clock,
    internalExecutor sqlutil.InternalExecutor,
    settings *cluster.Settings,
    testingKnobs LeaseManagerTestingKnobs,
    stopper *stop.Stopper,
    cfg *base.LeaseManagerConfig,
) *LeaseManager

NewLeaseManager creates a new LeaseManager.

internalExecutor can be nil to help bootstrapping, but then it needs to be set via SetInternalExecutor before the LeaseManager is used.

stopper is used to run async tasks. Can be nil in tests.

func (*LeaseManager) Acquire Uses

func (m *LeaseManager) Acquire(
    ctx context.Context, timestamp hlc.Timestamp, tableID sqlbase.ID,
) (*sqlbase.ImmutableTableDescriptor, hlc.Timestamp, error)

Acquire acquires a read lease for the specified table ID valid for the timestamp. It returns the table descriptor and a expiration time. A transaction using this descriptor must ensure that its commit-timestamp < expiration-time. Care must be taken to not modify the returned descriptor.

Known limitation: Acquire() can return an error after the table with the tableID has been dropped. This is true even when using a timestamp less than the timestamp of the DROP command. This is because Acquire can only return an older version of a descriptor if the latest version can be leased; as it stands a dropped table cannot be leased.

func (*LeaseManager) AcquireByName Uses

func (m *LeaseManager) AcquireByName(
    ctx context.Context, timestamp hlc.Timestamp, dbID sqlbase.ID, tableName string,
) (*sqlbase.ImmutableTableDescriptor, hlc.Timestamp, error)

AcquireByName returns a table version for the specified table valid for the timestamp. It returns the table descriptor and a expiration time. A transaction using this descriptor must ensure that its commit-timestamp < expiration-time. Care must be taken to not modify the returned descriptor. Renewal of a lease may begin in the background. Renewal is done in order to prevent blocking on future acquisitions.

Known limitation: AcquireByName() calls Acquire() and therefore suffers from the same limitation as Acquire (See Acquire). AcquireByName() is unable to function correctly on a timestamp less than the timestamp of a transaction with a DROP/TRUNCATE on a table. The limitation in the face of a DROP follows directly from the limitation on Acquire(). A TRUNCATE is implemented by changing the name -> id mapping for a table and by dropping the descriptor with the old id. While AcquireByName can use the timestamp and get the correct name->id mapping at a timestamp, it uses Acquire() to get a descriptor with the corresponding id and fails because the id has been dropped by the TRUNCATE.

func (*LeaseManager) AcquireFreshestFromStore Uses

func (m *LeaseManager) AcquireFreshestFromStore(ctx context.Context, tableID sqlbase.ID) error

AcquireFreshestFromStore acquires a new lease from the store and inserts it into the active set. It guarantees that the lease returned is the one acquired after the call is made. Use this if the lease we want to get needs to see some descriptor updates that we know happened recently.

func (*LeaseManager) DeleteOrphanedLeases Uses

func (m *LeaseManager) DeleteOrphanedLeases(timeThreshold int64)

DeleteOrphanedLeases releases all orphaned leases created by a prior instance of this node. timeThreshold is a walltime lower than the lowest hlc timestamp that the current instance of the node can use.

func (*LeaseManager) PeriodicallyRefreshSomeLeases Uses

func (m *LeaseManager) PeriodicallyRefreshSomeLeases()

PeriodicallyRefreshSomeLeases so that leases are fresh and can serve traffic immediately. TODO(vivek): Remove once epoch based table leases are implemented.

func (*LeaseManager) RefreshLeases Uses

func (m *LeaseManager) RefreshLeases(s *stop.Stopper, db *client.DB, g *gossip.Gossip)

RefreshLeases starts a goroutine that refreshes the lease manager leases for tables received in the latest system configuration via gossip.

func (*LeaseManager) Release Uses

func (m *LeaseManager) Release(desc *sqlbase.ImmutableTableDescriptor) error

Release releases a previously acquired table.

func (*LeaseManager) SetDraining Uses

func (m *LeaseManager) SetDraining(drain bool)

SetDraining (when called with 'true') removes all inactive leases. Any leases that are active will be removed once the lease's reference count drops to 0.

func (*LeaseManager) SetInternalExecutor Uses

func (m *LeaseManager) SetInternalExecutor(executor sqlutil.InternalExecutor)

SetInternalExecutor has to be called if a nil execCfg was passed to NewLeaseManager.

type LeaseManagerTestingKnobs Uses

type LeaseManagerTestingKnobs struct {
    // A callback called when a gossip update is received, before the leases are
    // refreshed. Careful when using this to block for too long - you can block
    // all the gossip users in the system. If it returns an error the gossip
    // update is ignored.
    GossipUpdateEvent func(*config.SystemConfig) error
    // A callback called after the leases are refreshed as a result of a gossip update.
    TestingLeasesRefreshedEvent func(*config.SystemConfig)
    // To disable the deletion of orphaned leases at server startup.
    DisableDeleteOrphanedLeases bool
    LeaseStoreTestingKnobs      LeaseStoreTestingKnobs
}

LeaseManagerTestingKnobs contains test knobs.

func (*LeaseManagerTestingKnobs) ModuleTestingKnobs Uses

func (*LeaseManagerTestingKnobs) ModuleTestingKnobs()

ModuleTestingKnobs is part of the base.ModuleTestingKnobs interface.

type LeaseStore Uses

type LeaseStore struct {
    // contains filtered or unexported fields
}

LeaseStore implements the operations for acquiring and releasing leases and publishing a new version of a descriptor. Exported only for testing.

func (LeaseStore) Publish Uses

func (s LeaseStore) Publish(
    ctx context.Context,
    tableID sqlbase.ID,
    update func(*sqlbase.MutableTableDescriptor) error,
    logEvent func(*client.Txn) error,
) (*sqlbase.ImmutableTableDescriptor, error)

Publish updates a table descriptor. It also maintains the invariant that there are at most two versions of the descriptor out in the wild at any time by first waiting for all nodes to be on the current (pre-update) version of the table desc.

The update closure is called after the wait, and it provides the new version of the descriptor to be written. In a multi-step schema operation, this update should perform a single step.

The closure may be called multiple times if retries occur; make sure it does not have side effects.

Returns the updated version of the descriptor.

func (LeaseStore) PublishMultiple Uses

func (s LeaseStore) PublishMultiple(
    ctx context.Context,
    tableIDs []sqlbase.ID,
    update func(map[sqlbase.ID]*sqlbase.MutableTableDescriptor) error,
    logEvent func(*client.Txn) error,
) (map[sqlbase.ID]*sqlbase.ImmutableTableDescriptor, error)

PublishMultiple updates multiple table descriptors, maintaining the invariant that there are at most two versions of each descriptor out in the wild at any time by first waiting for all nodes to be on the current (pre-update) version of the table desc.

The update closure for all tables is called after the wait. The argument to the closure is a map of the table descriptors with the IDs given in tableIDs, and the closure mutates those descriptors.

The closure may be called multiple times if retries occur; make sure it does not have side effects.

Returns the updated versions of the descriptors.

func (LeaseStore) WaitForOneVersion Uses

func (s LeaseStore) WaitForOneVersion(
    ctx context.Context, tableID sqlbase.ID, retryOpts retry.Options,
) (sqlbase.DescriptorVersion, error)

WaitForOneVersion returns once there are no unexpired leases on the previous version of the table descriptor. It returns the current version. After returning there can only be versions of the descriptor >= to the returned version. Lease acquisition (see acquire()) maintains the invariant that no new leases for desc.Version-1 will be granted once desc.Version exists.

type LeaseStoreTestingKnobs Uses

type LeaseStoreTestingKnobs struct {
    // Called after a lease is removed from the store, with any operation error.
    // See LeaseRemovalTracker.
    LeaseReleasedEvent func(id sqlbase.ID, version sqlbase.DescriptorVersion, err error)
    // Called after a lease is acquired, with any operation error.
    LeaseAcquiredEvent func(table sqlbase.TableDescriptor, err error)
    // Called before waiting on a results from a DoChan call of acquireNodeLease
    // in tableState.acquire() and tableState.acquireFreshestFromStore().
    LeaseAcquireResultBlockEvent func(leaseBlockType LeaseAcquireBlockType)
    // RemoveOnceDereferenced forces leases to be removed
    // as soon as they are dereferenced.
    RemoveOnceDereferenced bool
}

LeaseStoreTestingKnobs contains testing knobs.

func (*LeaseStoreTestingKnobs) ModuleTestingKnobs Uses

func (*LeaseStoreTestingKnobs) ModuleTestingKnobs()

ModuleTestingKnobs is part of the base.ModuleTestingKnobs interface.

type LogicalSchemaAccessor Uses

type LogicalSchemaAccessor struct {
    SchemaAccessor
    // contains filtered or unexported fields
}

LogicalSchemaAccessor extends an existing DatabaseLister with the ability to list tables in a virtual schema.

func (*LogicalSchemaAccessor) GetObjectDesc Uses

func (l *LogicalSchemaAccessor) GetObjectDesc(
    ctx context.Context, txn *client.Txn, name *ObjectName, flags tree.ObjectLookupFlags,
) (ObjectDescriptor, error)

GetObjectDesc implements the ObjectAccessor interface.

func (*LogicalSchemaAccessor) GetObjectNames Uses

func (l *LogicalSchemaAccessor) GetObjectNames(
    ctx context.Context,
    txn *client.Txn,
    dbDesc *DatabaseDescriptor,
    scName string,
    flags tree.DatabaseListFlags,
) (TableNames, error)

GetObjectNames implements the DatabaseLister interface.

func (*LogicalSchemaAccessor) IsValidSchema Uses

func (l *LogicalSchemaAccessor) IsValidSchema(dbDesc *DatabaseDescriptor, scName string) bool

IsValidSchema implements the DatabaseLister interface.

type MemoryMetrics Uses

type MemoryMetrics struct {
    MaxBytesHist         *metric.Histogram
    CurBytesCount        *metric.Gauge
    TxnMaxBytesHist      *metric.Histogram
    TxnCurBytesCount     *metric.Gauge
    SessionMaxBytesHist  *metric.Histogram
    SessionCurBytesCount *metric.Gauge
}

MemoryMetrics contains pointers to the metrics object for one of the SQL endpoints: - "client" for connections received via pgwire. - "admin" for connections received via the admin RPC. - "internal" for activities related to leases, schema changes, etc.

func MakeMemMetrics Uses

func MakeMemMetrics(endpoint string, histogramWindow time.Duration) MemoryMetrics

MakeMemMetrics instantiates the metric objects for an SQL endpoint.

func (MemoryMetrics) MetricStruct Uses

func (MemoryMetrics) MetricStruct()

MetricStruct implements the metrics.Struct interface.

type Metrics Uses

type Metrics struct {
    // EngineMetrics is exported as required by the metrics.Struct magic we use
    // for metrics registration.
    EngineMetrics EngineMetrics

    // StartedStatementCounters contains metrics for statements initiated by
    // users. These metrics count user-initiated operations, regardless of
    // success (in particular, TxnCommitCount is the number of COMMIT statements
    // attempted, not the number of transactions that successfully commit).
    StartedStatementCounters StatementCounters

    // ExecutedStatementCounters contains metrics for successfully executed
    // statements.
    ExecutedStatementCounters StatementCounters
}

Metrics collects timeseries data about SQL activity.

type MutableTableDescriptor Uses

type MutableTableDescriptor = sqlbase.MutableTableDescriptor

MutableTableDescriptor is provided for convenience and to make the interface definitions below more intuitive.

func ResolveMutableExistingObject Uses

func ResolveMutableExistingObject(
    ctx context.Context,
    sc SchemaResolver,
    tn *ObjectName,
    required bool,
    requiredType ResolveRequiredType,
) (res *MutableTableDescriptor, err error)

ResolveMutableExistingObject looks up an existing mutable object. If required is true, an error is returned if the object does not exist. Optionally, if a desired descriptor type is specified, that type is checked.

The object name is modified in-place with the result of the name resolution, if successful. It is not modified in case of error or if no object is found.

type NodeInfo Uses

type NodeInfo struct {
    ClusterID func() uuid.UUID
    NodeID    *base.NodeIDContainer
    AdminURL  func() *url.URL
    PGURL     func(*url.Userinfo) (*url.URL, error)
}

NodeInfo contains metadata about the executing node and cluster.

type ObjectDescriptor Uses

type ObjectDescriptor interface {
    tree.NameResolutionResult

    // TableDesc returns the underlying table descriptor.
    TableDesc() *TableDescriptor
}

ObjectDescriptor provides table information for results from a name lookup.

type ObjectName Uses

type ObjectName = tree.TableName

ObjectName is provided for convenience and to make the interface definitions below more intuitive.

type PGWireTestingKnobs Uses

type PGWireTestingKnobs struct {
    // CatchPanics causes the pgwire.conn to recover from panics in its execution
    // thread and return them as errors to the client, closing the connection
    // afterward.
    CatchPanics bool

    // AuthHook is used to override the normal authentication handling on new
    // connections.
    AuthHook func(context.Context) error
}

PGWireTestingKnobs contains knobs for the pgwire module.

func (*PGWireTestingKnobs) ModuleTestingKnobs Uses

func (*PGWireTestingKnobs) ModuleTestingKnobs()

ModuleTestingKnobs implements the base.ModuleTestingKnobs interface.

type ParseResult Uses

type ParseResult interface {
    ResultBase
}

ParseResult represents the result of a Parse command.

type PhysicalPlan Uses

type PhysicalPlan struct {
    physicalplan.PhysicalPlan

    // PlanToStreamColMap maps planNode columns (see planColumns()) to columns in
    // the result streams. These stream indices correspond to the streams
    // referenced in ResultTypes.
    //
    // Note that in some cases, not all columns in the result streams are
    // referenced in the map; for example, columns that are only required for
    // stream merges in downstream input synchronizers are not included here.
    // (This is due to some processors not being configurable to output only
    // certain columns and will be fixed.)
    //
    // Conversely, in some cases not all planNode columns have a corresponding
    // result stream column (these map to index -1); this is the case for scanNode
    // and indexJoinNode where not all columns in the table are actually used in
    // the plan, but are kept for possible use downstream (e.g., sorting).
    //
    // When the query is run, the output processor's PlanToStreamColMap is used
    // by DistSQLReceiver to create an implicit projection on the processor's
    // output for client consumption (see DistSQLReceiver.Push()). Therefore,
    // "invisible" columns (e.g., columns required for merge ordering) will not
    // be output.
    PlanToStreamColMap []int
}

PhysicalPlan is a partial physical plan which corresponds to a planNode (partial in that it can correspond to a planNode subtree and not necessarily to the entire planNode for a given query).

It augments physicalplan.PhysicalPlan with information relating the physical plan to a planNode subtree.

These plans are built recursively on a planNode tree.

type PlanHookRowFn Uses

type PlanHookRowFn func(context.Context, []planNode, chan<- tree.Datums) error

PlanHookRowFn describes the row-production for hook-created plans. The channel argument is used to return results to the plan's runner. It's a blocking channel, so implementors should be careful to only use blocking sends on it when necessary. Any subplans returned by the hook when initially called are passed back, planned and started, for the the RowFn's use.

TODO(dt): should this take runParams like a normal planNode.Next?

type PlanHookState Uses

type PlanHookState interface {
    SchemaResolver
    RunParams(ctx context.Context) runParams
    ExtendedEvalContext() *extendedEvalContext
    SessionData() *sessiondata.SessionData
    ExecCfg() *ExecutorConfig
    DistSQLPlanner() *DistSQLPlanner
    LeaseMgr() *LeaseManager
    TypeAsString(e tree.Expr, op string) (func() (string, error), error)
    TypeAsStringArray(e tree.Exprs, op string) (func() ([]string, error), error)
    TypeAsStringOpts(
        opts tree.KVOptions, optsValidate map[string]KVStringOptValidate,
    ) (func() (map[string]string, error), error)
    User() string
    AuthorizationAccessor
    // The role create/drop call into OSS code to reuse plan nodes.
    // TODO(mberhault): it would be easier to just pass a planner to plan hooks.
    CreateUserNode(
        ctx context.Context, nameE, passwordE tree.Expr, ifNotExists bool, isRole bool, opName string,
    ) (*CreateUserNode, error)
    DropUserNode(
        ctx context.Context, namesE tree.Exprs, ifExists bool, isRole bool, opName string,
    ) (*DropUserNode, error)
    GetAllUsersAndRoles(ctx context.Context) (map[string]bool, error)
    BumpRoleMembershipTableVersion(ctx context.Context) error
    EvalAsOfTimestamp(asOf tree.AsOfClause) (hlc.Timestamp, error)
    ResolveUncachedDatabaseByName(
        ctx context.Context, dbName string, required bool) (*UncachedDatabaseDescriptor, error)
    ResolveMutableTableDescriptor(
        ctx context.Context, tn *ObjectName, required bool, requiredType ResolveRequiredType,
    ) (table *MutableTableDescriptor, err error)
}

PlanHookState exposes the subset of planner needed by plan hooks. We pass this as one interface, rather than individually passing each field or interface as we find we need them, to avoid churn in the planHookFn sig and the hooks that implement it.

The PlanHookState is used by modules that are under the CCL. Since the OSS modules cannot depend on the CCL modules, the CCL modules need to inform the planner when they should be invoked (via plan hooks). The only way for the CCL statements to get access to a "planner" is through this PlanHookState that gets passed back due to this inversion of roles.

type PlanNode Uses

type PlanNode = planNode

PlanNode is the exported name for planNode. Useful for CCL hooks.

type PlanningCtx Uses

type PlanningCtx struct {
    ExtendedEvalCtx *extendedEvalContext

    // NodeAddresses contains addresses for all NodeIDs that are referenced by any
    // PhysicalPlan we generate with this context.
    // Nodes that fail a health check have empty addresses.
    NodeAddresses map[roachpb.NodeID]string
    // contains filtered or unexported fields
}

PlanningCtx contains data used and updated throughout the planning process of a single query.

func (*PlanningCtx) EvalContext Uses

func (p *PlanningCtx) EvalContext() *tree.EvalContext

EvalContext returns the associated EvalContext, or nil if there isn't one.

func (*PlanningCtx) EvaluateSubqueries Uses

func (p *PlanningCtx) EvaluateSubqueries() bool

EvaluateSubqueries returns true if this plan requires subqueries be fully executed before trying to marshal. This is normally true except for in the case of EXPLAIN queries, which ultimately want to describe the subquery that will run, without actually running it.

func (*PlanningCtx) IsLocal Uses

func (p *PlanningCtx) IsLocal() bool

IsLocal returns true if this PlanningCtx is being used to plan a query that has no remote flows.

type PrepareStmt Uses

type PrepareStmt struct {
    // Name of the prepared statement (optional).
    Name string

    // Information returned from parsing: AST, SQL, NumPlaceholders.
    // Note that AST can be nil, in which case executing it should produce an
    // "empty query response" message.
    parser.Statement

    TypeHints tree.PlaceholderTypes
    // RawTypeHints is the representation of type hints exactly as specified by
    // the client.
    RawTypeHints []oid.Oid
    ParseStart   time.Time
    ParseEnd     time.Time
}

PrepareStmt is the command for creating a prepared statement.

func (PrepareStmt) String Uses

func (p PrepareStmt) String() string

type PreparedPortal Uses

type PreparedPortal struct {
    Stmt  *PreparedStatement
    Qargs tree.QueryArguments

    // OutFormats contains the requested formats for the output columns.
    OutFormats []pgwirebase.FormatCode
    // contains filtered or unexported fields
}

PreparedPortal is a PreparedStatement that has been bound with query arguments.

Note that PreparedPortals maintain a reference counter internally. References need to be registered with incRef() and de-registered with decRef().

type PreparedStatement Uses

type PreparedStatement struct {
    sqlbase.PrepareMetadata

    // Memo is the memoized data structure constructed by the cost-based optimizer
    // during prepare of a SQL statement. It can significantly speed up execution
    // if it is used by the optimizer as a starting point.
    Memo *memo.Memo
    // contains filtered or unexported fields
}

PreparedStatement is a SQL statement that has been parsed and the types of arguments and results have been determined.

Note that PreparedStatemts maintain a reference counter internally. References need to be registered with incRef() and de-registered with decRef().

func (*PreparedStatement) MemoryEstimate Uses

func (p *PreparedStatement) MemoryEstimate() int64

MemoryEstimate returns a rough estimate of the PreparedStatement's memory usage, in bytes.

type PreparedStatementOrigin Uses

type PreparedStatementOrigin int

PreparedStatementOrigin is an enum representing the source of where the prepare statement was made.

const (
    // PreparedStatementOriginWire signifies the prepared statement was made
    // over the wire.
    PreparedStatementOriginWire PreparedStatementOrigin = iota + 1
    // PreparedStatementOriginSQL signifies the prepared statement was made
    // over a parsed SQL query.
    PreparedStatementOriginSQL
)

type ReqOrdering Uses

type ReqOrdering = sqlbase.ColumnOrdering

ReqOrdering is the ordering that must be preserved by an operator when it is distributed. It is used to configure DistSQL with the orderings it needs to maintain when joining streams.

type ResolveRequiredType Uses

type ResolveRequiredType int

ResolveRequiredType can be passed to the ResolveExistingObject function to require the returned descriptor to be of a specific type.

const (
    ResolveAnyDescType ResolveRequiredType = iota
    ResolveRequireTableDesc
    ResolveRequireViewDesc
    ResolveRequireTableOrViewDesc
    ResolveRequireSequenceDesc
)

ResolveRequiredType options have descriptive names.

type RestrictedCommandResult Uses

type RestrictedCommandResult interface {
    CommandResultErrBase

    // SetColumns informs the client about the schema of the result. The columns
    // can be nil.
    //
    // This needs to be called (once) before AddRow.
    SetColumns(context.Context, sqlbase.ResultColumns)

    // ResetStmtType allows a client to change the statement type of the current
    // result, from the original one set when the result was created trough
    // ClientComm.createStatementResult.
    ResetStmtType(stmt tree.Statement)

    // AddRow accumulates a result row.
    //
    // The implementation cannot hold on to the row slice; it needs to make a
    // shallow copy if it needs to.
    AddRow(ctx context.Context, row tree.Datums) error

    // IncrementRowsAffected increments a counter by n. This is used for all
    // result types other than tree.Rows.
    IncrementRowsAffected(n int)

    // RowsAffected returns either the number of times AddRow was called, or the
    // sum of all n passed into IncrementRowsAffected.
    RowsAffected() int

    // DisableBuffering can be called during execution to ensure that
    // the results accumulated so far, and all subsequent rows added
    // to this CommandResult, will be flushed immediately to the client.
    // This is currently used for sinkless changefeeds.
    DisableBuffering()
}

RestrictedCommandResult is a subset of CommandResult meant to make it clear that its clients don't close the CommandResult.

type ResultBase Uses

type ResultBase interface {
    CommandResultErrBase
    CommandResultClose
}

ResultBase is the common interface implemented by all the different command results.

type RowDescOpt Uses

type RowDescOpt bool

RowDescOpt specifies whether a result needs a row description message.

const (
    // NeedRowDesc specifies that a row description message is needed.
    NeedRowDesc RowDescOpt = false
    // DontNeedRowDesc specifies that a row description message is not needed.
    DontNeedRowDesc RowDescOpt = true
)

type RowResultWriter Uses

type RowResultWriter struct {
    // contains filtered or unexported fields
}

RowResultWriter is a thin wrapper around a RowContainer.

func NewRowResultWriter Uses

func NewRowResultWriter(rowContainer *rowcontainer.RowContainer) *RowResultWriter

NewRowResultWriter creates a new RowResultWriter.

func (*RowResultWriter) AddRow Uses

func (b *RowResultWriter) AddRow(ctx context.Context, row tree.Datums) error

AddRow implements the rowResultWriter interface.

func (*RowResultWriter) Err Uses

func (b *RowResultWriter) Err() error

Err is part of the rowResultWriter interface.

func (*RowResultWriter) IncrementRowsAffected Uses

func (b *RowResultWriter) IncrementRowsAffected(n int)

IncrementRowsAffected implements the rowResultWriter interface.

func (*RowResultWriter) SetError Uses

func (b *RowResultWriter) SetError(err error)

SetError is part of the rowResultWriter interface.

type SchemaAccessor Uses

type SchemaAccessor interface {
    // GetDatabaseDesc looks up a database by name and returns its
    // descriptor. If the database is not found and required is true,
    // an error is returned; otherwise a nil reference is returned.
    GetDatabaseDesc(ctx context.Context, txn *client.Txn, dbName string, flags tree.DatabaseLookupFlags) (*DatabaseDescriptor, error)

    // IsValidSchema returns true if the given schema name is valid for the given database.
    IsValidSchema(db *DatabaseDescriptor, scName string) bool

    // GetObjectNames returns the list of all objects in the given
    // database and schema.
    // TODO(whomever): when separate schemas are supported, this
    // API should be extended to use schema descriptors.
    GetObjectNames(ctx context.Context, txn *client.Txn, db *DatabaseDescriptor, scName string, flags tree.DatabaseListFlags) (TableNames, error)

    // GetObjectDesc looks up an object by name and returns both its
    // descriptor and that of its parent database. If the object is not
    // found and flags.required is true, an error is returned, otherwise
    // a nil reference is returned.
    GetObjectDesc(ctx context.Context, txn *client.Txn, name *ObjectName, flags tree.ObjectLookupFlags) (ObjectDescriptor, error)
}

SchemaAccessor provides access to database descriptors.

type SchemaChangeManager Uses

type SchemaChangeManager struct {
    // contains filtered or unexported fields
}

SchemaChangeManager processes pending schema changes seen in gossip updates. Most schema changes are executed synchronously by the node that created the schema change. If the node dies while processing the schema change this manager acts as a backup execution mechanism.

func NewSchemaChangeManager Uses

func NewSchemaChangeManager(
    ambientCtx log.AmbientContext,
    execCfg *ExecutorConfig,
    testingKnobs *SchemaChangerTestingKnobs,
    db client.DB,
    nodeDesc roachpb.NodeDescriptor,
    dsp *DistSQLPlanner,
    ieFactory sqlutil.SessionBoundInternalExecutorFactory,
) *SchemaChangeManager

NewSchemaChangeManager returns a new SchemaChangeManager.

func (*SchemaChangeManager) Start Uses

func (s *SchemaChangeManager) Start(stopper *stop.Stopper)

Start starts a goroutine that runs outstanding schema changes for tables received in the latest system configuration via gossip.

type SchemaChanger Uses

type SchemaChanger struct {
    // contains filtered or unexported fields
}

SchemaChanger is used to change the schema on a table.

func NewSchemaChangerForTesting Uses

func NewSchemaChangerForTesting(
    tableID sqlbase.ID,
    mutationID sqlbase.MutationID,
    nodeID roachpb.NodeID,
    db client.DB,
    leaseMgr *LeaseManager,
    jobRegistry *jobs.Registry,
    execCfg *ExecutorConfig,
    settings *cluster.Settings,
) SchemaChanger

NewSchemaChangerForTesting only for tests.

func (*SchemaChanger) AcquireLease Uses

func (sc *SchemaChanger) AcquireLease(
    ctx context.Context,
) (sqlbase.TableDescriptor_SchemaChangeLease, error)

AcquireLease acquires a schema change lease on the table if an unexpired lease doesn't exist. It returns the lease.

func (*SchemaChanger) DropTableDesc Uses

func (sc *SchemaChanger) DropTableDesc(
    ctx context.Context, tableDesc *sqlbase.TableDescriptor, traceKV bool,
) error

DropTableDesc removes a descriptor from the KV database.

func (*SchemaChanger) ExtendLease Uses

func (sc *SchemaChanger) ExtendLease(
    ctx context.Context, existingLease *sqlbase.TableDescriptor_SchemaChangeLease,
) error

ExtendLease for the current leaser. This needs to be called often while doing a schema change to prevent more than one node attempting to apply a schema change (which is still safe, but unwise). It updates existingLease with the new lease.

func (*SchemaChanger) ReleaseLease Uses

func (sc *SchemaChanger) ReleaseLease(
    ctx context.Context, lease sqlbase.TableDescriptor_SchemaChangeLease,
) error

ReleaseLease releases the table lease if it is the one registered with the table descriptor.

func (*SchemaChanger) RunStateMachineBeforeBackfill Uses

func (sc *SchemaChanger) RunStateMachineBeforeBackfill(ctx context.Context) error

RunStateMachineBeforeBackfill moves the state machine forward and wait to ensure that all nodes are seeing the latest version of the table.

type SchemaChangerTestingKnobs Uses

type SchemaChangerTestingKnobs struct {
    // SyncFilter is called before running schema changers synchronously (at
    // the end of a txn). The function can be used to clear the schema
    // changers (if the test doesn't want them run using the synchronous path)
    // or to temporarily block execution. Note that this has nothing to do
    // with the async path for running schema changers. To block that, set
    // AsyncExecNotification.
    SyncFilter SyncSchemaChangersFilter

    // RunBeforePublishWriteAndDelete is called just before publishing the
    // write+delete state for the schema change.
    RunBeforePublishWriteAndDelete func()

    // RunBeforeBackfill is called just before starting the backfill.
    RunBeforeBackfill func() error

    // RunBeforeIndexBackfill is called just before starting the index backfill, after
    // fixing the index backfill scan timestamp.
    RunBeforeIndexBackfill func()

    // RunBeforeIndexValidation is called just before starting the index validation,
    // after setting the job status to validating.
    RunBeforeIndexValidation func() error

    // RunBeforeConstraintValidation is called just before starting the checks validation,
    // after setting the job status to validating.
    RunBeforeConstraintValidation func() error

    // OldNamesDrainedNotification is called during a schema change,
    // after all leases on the version of the descriptor with the old
    // names are gone, and just before the mapping of the old names to the
    // descriptor id are about to be deleted.
    OldNamesDrainedNotification func()

    // AsyncExecNotification is a function called before running a schema
    // change asynchronously. Returning an error will prevent the asynchronous
    // execution path from running.
    AsyncExecNotification func() error

    // AsyncExecQuickly executes queued schema changes as soon as possible.
    AsyncExecQuickly bool

    // WriteCheckpointInterval is the interval after which a checkpoint is
    // written.
    WriteCheckpointInterval time.Duration

    // BackfillChunkSize is to be used for all backfill chunked operations.
    BackfillChunkSize int64

    // TwoVersionLeaseViolation is called whenever a schema change
    // transaction is unable to commit because it is violating the two
    // version lease invariant.
    TwoVersionLeaseViolation func()

    // OnError is called with all the errors seen by the
    // synchronous code path.
    OnError func(err error)
}

SchemaChangerTestingKnobs for testing the schema change execution path through both the synchronous and asynchronous paths.

func (*SchemaChangerTestingKnobs) ModuleTestingKnobs Uses

func (*SchemaChangerTestingKnobs) ModuleTestingKnobs()

ModuleTestingKnobs is part of the base.ModuleTestingKnobs interface.

type SchemaResolver Uses

type SchemaResolver interface {
    tree.TableNameExistingResolver
    tree.TableNameTargetResolver

    Txn() *client.Txn
    LogicalSchemaAccessor() SchemaAccessor
    CurrentDatabase() string
    CurrentSearchPath() sessiondata.SearchPath
    CommonLookupFlags(required bool) tree.CommonLookupFlags
    ObjectLookupFlags(required bool, requireMutable bool) tree.ObjectLookupFlags
    LookupTableByID(ctx context.Context, id sqlbase.ID) (row.TableEntry, error)
}

SchemaResolver abstracts the interfaces needed from the logical planner to perform name resolution below.

We use an interface instead of passing *planner directly to make the resolution methods able to work even when we evolve the code to use a different plan builder. TODO(rytaft,andyk): study and reuse this.

type SendError Uses

type SendError struct {
    // Err is a *pgerror.Error.
    Err error
}

SendError is a command that, upon execution, send a specific error to the client. This is used by pgwire to schedule errors to be sent at an appropriate time.

func (SendError) String Uses

func (s SendError) String() string

type SequenceDescriptor Uses

type SequenceDescriptor = sqlbase.TableDescriptor

SequenceDescriptor is provided for convenience and to make the interface definitions below more intuitive.

type Server Uses

type Server struct {

    // Metrics is used to account normal queries.
    Metrics Metrics

    // InternalMetrics is used to account internal queries.
    InternalMetrics Metrics
    // contains filtered or unexported fields
}

Server is the top level singleton for handling SQL connections. It creates connExecutors to server every incoming connection.

func NewServer Uses

func NewServer(cfg *ExecutorConfig, pool *mon.BytesMonitor) *Server

NewServer creates a new Server. Start() needs to be called before the Server is used.

func (*Server) GetExecutorConfig Uses

func (s *Server) GetExecutorConfig() *ExecutorConfig

GetExecutorConfig returns this server's executor config.

func (*Server) GetScrubbedStmtStats Uses

func (s *Server) GetScrubbedStmtStats() []roachpb.CollectedStatementStatistics

GetScrubbedStmtStats returns the statement statistics by app, with the queries scrubbed of their identifiers. Any statements which cannot be scrubbed will be omitted from the returned map.

func (*Server) GetStmtStatsLastReset Uses

func (s *Server) GetStmtStatsLastReset() time.Time

GetStmtStatsLastReset returns the time at which the statement statistics were last cleared.

func (*Server) GetUnscrubbedStmtStats Uses

func (s *Server) GetUnscrubbedStmtStats() []roachpb.CollectedStatementStatistics

GetUnscrubbedStmtStats returns the same thing as GetScrubbedStmtStats, except identifiers (e.g. table and column names) aren't scrubbed from the statements.

func (*Server) PeriodicallyClearSQLStats Uses

func (s *Server) PeriodicallyClearSQLStats(ctx context.Context, stopper *stop.Stopper)

PeriodicallyClearSQLStats runs a loop to ensure that sql stats are reset. Usually we expect those stats to be reset by diagnostics reporting, after it generates its reports. However if the diagnostics loop crashes and stops resetting stats, this loop ensures stats do not accumulate beyond a the diagnostics.forced_stat_reset.interval limit.

func (*Server) ResetSQLStats Uses

func (s *Server) ResetSQLStats(ctx context.Context)

ResetSQLStats resets the executor's collected sql statistics.

func (*Server) ServeConn Uses

func (s *Server) ServeConn(
    ctx context.Context, h ConnectionHandler, reserved mon.BoundAccount, cancel context.CancelFunc,
) error

ServeConn serves a client connection by reading commands from the stmtBuf embedded in the ConnHandler.

If not nil, reserved represents memory reserved for the connection. The connExecutor takes ownership of this memory.

func (*Server) SetupConn Uses

func (s *Server) SetupConn(
    ctx context.Context,
    args SessionArgs,
    stmtBuf *StmtBuf,
    clientComm ClientComm,
    memMetrics MemoryMetrics,
) (ConnectionHandler, error)

SetupConn creates a connExecutor for the client connection.

When this method returns there are no resources allocated yet that need to be close()d.

Args: args: The initial session parameters. They are validated by SetupConn

and an error is returned if this validation fails.

stmtBuf: The incoming statement for the new connExecutor. clientComm: The interface through which the new connExecutor is going to

produce results for the client.

memMetrics: The metrics that statements executed on this connection will

contribute to.

func (*Server) Start Uses

func (s *Server) Start(ctx context.Context, stopper *stop.Stopper)

Start starts the Server's background processing.

type SessionArgs Uses

type SessionArgs struct {
    User            string
    SessionDefaults SessionDefaults
    // RemoteAddr is the client's address. This is nil iff this is an internal
    // client.
    RemoteAddr            net.Addr
    ConnResultsBufferSize int64
}

SessionArgs contains arguments for serving a client connection.

type SessionBoundInternalExecutor Uses

type SessionBoundInternalExecutor struct {
    // contains filtered or unexported fields
}

SessionBoundInternalExecutor is like InternalExecutor, except that it is initialized with values for session variables. Conversely, it doesn't offer the *WithUser methods of the InternalExecutor.

func NewSessionBoundInternalExecutor Uses

func NewSessionBoundInternalExecutor(
    ctx context.Context,
    sessionData *sessiondata.SessionData,
    s *Server,
    memMetrics MemoryMetrics,
    settings *cluster.Settings,
) *SessionBoundInternalExecutor

NewSessionBoundInternalExecutor creates a SessionBoundInternalExecutor.

func (*SessionBoundInternalExecutor) Exec Uses

func (ie *SessionBoundInternalExecutor) Exec(
    ctx context.Context, opName string, txn *client.Txn, stmt string, qargs ...interface{},
) (int, error)

Exec executes the supplied SQL statement.

If txn is not nil, the statement will be executed in the respective txn.

Returns the number of rows affected.

func (*SessionBoundInternalExecutor) Query Uses

func (ie *SessionBoundInternalExecutor) Query(
    ctx context.Context, opName string, txn *client.Txn, stmt string, qargs ...interface{},
) ([]tree.Datums, error)

Query executes the supplied SQL statement and returns the resulting rows. The statement is executed as the root user.

If txn is not nil, the statement will be executed in the respective txn.

func (*SessionBoundInternalExecutor) QueryRow Uses

func (ie *SessionBoundInternalExecutor) QueryRow(
    ctx context.Context, opName string, txn *client.Txn, stmt string, qargs ...interface{},
) (tree.Datums, error)

QueryRow is like Query, except it returns a single row, or nil if not row is found, or an error if more that one row is returned.

func (*SessionBoundInternalExecutor) QueryWithCols Uses

func (ie *SessionBoundInternalExecutor) QueryWithCols(
    ctx context.Context, opName string, txn *client.Txn, stmt string, qargs ...interface{},
) ([]tree.Datums, sqlbase.ResultColumns, error)

QueryWithCols is like Query, but it also returns the computed ResultColumns of the input query.

type SessionDefaults Uses

type SessionDefaults map[string]string

SessionDefaults mirrors fields in Session, for restoring default configuration values in SET ... TO DEFAULT (or RESET ...) statements.

type SessionRegistry Uses

type SessionRegistry struct {
    syncutil.Mutex
    // contains filtered or unexported fields
}

SessionRegistry stores a set of all sessions on this node. Use register() and deregister() to modify this registry.

func NewSessionRegistry Uses

func NewSessionRegistry() *SessionRegistry

NewSessionRegistry creates a new SessionRegistry with an empty set of sessions.

func (*SessionRegistry) CancelQuery Uses

func (r *SessionRegistry) CancelQuery(queryIDStr string, username string) (bool, error)

CancelQuery looks up the associated query in the session registry and cancels it.

func (*SessionRegistry) CancelSession Uses

func (r *SessionRegistry) CancelSession(sessionIDBytes []byte, username string) (bool, error)

CancelSession looks up the specified session in the session registry and cancels it.

func (*SessionRegistry) SerializeAll Uses

func (r *SessionRegistry) SerializeAll() []serverpb.Session

SerializeAll returns a slice of all sessions in the registry, converted to serverpb.Sessions.

type SessionTracing Uses

type SessionTracing struct {
    // contains filtered or unexported fields
}

SessionTracing holds the state used by SET TRACING {ON,OFF,LOCAL} statements in the context of one SQL session. It holds the current trace being collected (or the last trace collected, if tracing is not currently ongoing).

SessionTracing and its interactions with the connExecutor are thread-safe; tracing can be turned on at any time.

func (*SessionTracing) Enabled Uses

func (st *SessionTracing) Enabled() bool

Enabled checks whether session tracing is currently enabled.

func (*SessionTracing) KVTracingEnabled Uses

func (st *SessionTracing) KVTracingEnabled() bool

KVTracingEnabled checks whether KV tracing is currently enabled.

func (*SessionTracing) RecordingType Uses

func (st *SessionTracing) RecordingType() tracing.RecordingType

RecordingType returns which type of tracing is currently being done.

func (*SessionTracing) StartTracing Uses

func (st *SessionTracing) StartTracing(
    recType tracing.RecordingType, kvTracingEnabled, showResults bool,
) error

StartTracing starts "session tracing". From this moment on, everything happening on both the connection's context and the current txn's context (if any) will be traced. StopTracing() needs to be called to finish this trace.

There's two contexts on which we must record: 1) If we're inside a txn, we start recording on the txn's span. We assume that the txn's ctx has a recordable span on it. 2) Regardless of whether we're in a txn or not, we need to record the connection's context. This context generally does not have a span, so we "hijack" it with one that does. Whatever happens on that context, plus whatever happens in future derived txn contexts, will be recorded.

Args: kvTracingEnabled: If set, the traces will also include "KV trace" messages -

verbose messages around the interaction of SQL with KV. Some of the messages
are per-row.

showResults: If set, result rows are reported in the trace.

func (*SessionTracing) StopTracing Uses

func (st *SessionTracing) StopTracing() error

StopTracing stops the trace that was started with StartTracing(). An error is returned if tracing was not active.

func (*SessionTracing) TraceExecConsume Uses

func (st *SessionTracing) TraceExecConsume(ctx context.Context) (context.Context, func())

TraceExecConsume creates a context for TraceExecRowsResult below.

func (*SessionTracing) TraceExecEnd Uses

func (st *SessionTracing) TraceExecEnd(ctx context.Context, err error, count int)

TraceExecEnd conditionally emits a trace message at the moment plan execution completes.

func (*SessionTracing) TraceExecRowsResult Uses

func (st *SessionTracing) TraceExecRowsResult(ctx context.Context, values tree.Datums)

TraceExecRowsResult conditionally emits a trace message for a single output row.

func (*SessionTracing) TraceExecStart Uses

func (st *SessionTracing) TraceExecStart(ctx context.Context, engine string)

TraceExecStart conditionally emits a trace message at the moment plan execution starts.

func (*SessionTracing) TracePlanCheckEnd Uses

func (st *SessionTracing) TracePlanCheckEnd(ctx context.Context, err error, dist bool)

TracePlanCheckEnd conditionally emits a trace message at the moment the engine check ends.

func (*SessionTracing) TracePlanCheckStart Uses

func (st *SessionTracing) TracePlanCheckStart(ctx context.Context)

TracePlanCheckStart conditionally emits a trace message at the moment the test of which execution engine to use starts.

func (*SessionTracing) TracePlanEnd Uses

func (st *SessionTracing) TracePlanEnd(ctx context.Context, err error)

TracePlanEnd conditionally emits a trace message at the moment logical planning ends.

func (*SessionTracing) TracePlanStart Uses

func (st *SessionTracing) TracePlanStart(ctx context.Context, stmtTag string)

TracePlanStart conditionally emits a trace message at the moment logical planning starts.

type SpanPartition Uses

type SpanPartition struct {
    Node  roachpb.NodeID
    Spans roachpb.Spans
}

SpanPartition is the intersection between a set of spans for a certain operation (e.g table scan) and the set of ranges owned by a given node.

type Statement Uses

type Statement struct {
    parser.Statement

    ExpectedTypes sqlbase.ResultColumns
    AnonymizedStr string

    // Prepared is non-nil during the PREPARE phase, as well as during EXECUTE of
    // a previously prepared statement. The Prepared statement can be modified
    // during either phase; the PREPARE phase sets its initial state, and the
    // EXECUTE phase can re-prepare it. This happens when the original plan has
    // been invalidated by schema changes, session data changes, permission
    // changes, or other changes to the context in which the original plan was
    // prepared.
    //
    // Given that the PreparedStatement can be modified during planning, it is
    // not safe for use on multiple threads.
    Prepared *PreparedStatement
    // contains filtered or unexported fields
}

Statement contains a statement with optional expected result columns and metadata.

func (Statement) String Uses

func (s Statement) String() string

type StatementCounters Uses

type StatementCounters struct {
    // QueryCount includes all statements and it is therefore the sum of
    // all the below metrics.
    QueryCount telemetry.CounterWithMetric

    // Basic CRUD statements.
    SelectCount telemetry.CounterWithMetric
    UpdateCount telemetry.CounterWithMetric
    InsertCount telemetry.CounterWithMetric
    DeleteCount telemetry.CounterWithMetric

    // Transaction operations.
    TxnBeginCount    telemetry.CounterWithMetric
    TxnCommitCount   telemetry.CounterWithMetric
    TxnRollbackCount telemetry.CounterWithMetric

    // Savepoint operations. SavepointCount is for real SQL savepoints
    // (which we don't yet support; this is just a placeholder for
    // telemetry); the RestartSavepoint variants are for the
    // cockroach-specific client-side retry protocol.
    SavepointCount                  telemetry.CounterWithMetric
    RestartSavepointCount           telemetry.CounterWithMetric
    ReleaseRestartSavepointCount    telemetry.CounterWithMetric
    RollbackToRestartSavepointCount telemetry.CounterWithMetric

    // DdlCount counts all statements whose StatementType is DDL.
    DdlCount telemetry.CounterWithMetric

    // MiscCount counts all statements not covered by a more specific stat above.
    MiscCount telemetry.CounterWithMetric
}

StatementCounters groups metrics for counting different types of statements.

type StatementFilter Uses

type StatementFilter func(context.Context, string, error)

StatementFilter is the type of callback that ExecutorTestingKnobs.StatementFilter takes.

type StmtBuf Uses

type StmtBuf struct {
    // contains filtered or unexported fields
}

StmtBuf maintains a list of commands that a SQL client has sent for execution over a network connection. The commands are SQL queries to be executed, statements to be prepared, etc. At any point in time the buffer contains outstanding commands that have yet to be executed, and it can also contain some history of commands that we might want to retry - in the case of a retriable error, we'd like to retry all the commands pertaining to the current SQL transaction.

The buffer is supposed to be used by one reader and one writer. The writer adds commands to the buffer using Push(). The reader reads one command at a time using CurCmd(). The consumer is then supposed to create command results (the buffer is not involved in this). The buffer internally maintains a cursor representing the reader's position. The reader has to manually move the cursor using AdvanceOne(), seekToNextBatch() and rewind(). In practice, the writer is a module responsible for communicating with a SQL client (i.e. pgwire.conn) and the reader is a connExecutor.

The StmtBuf supports grouping commands into "batches" delimited by sync commands. A reader can then at any time chose to skip over commands from the current batch. This is used to implement Postgres error semantics: when an error happens during processing of a command, some future commands might need to be skipped. Batches correspond either to multiple queries received in a single query string (when the SQL client sends a semicolon-separated list of queries as part of the "simple" protocol), or to different commands pipelined by the cliend, separated from "sync" messages.

push() can be called concurrently with CurCmd().

The connExecutor will use the buffer to maintain a window around the command it is currently executing. It will maintain enough history for executing commands again in case of an automatic retry. The connExecutor is in charge of trimming completed commands from the buffer when it's done with them.

func NewStmtBuf Uses

func NewStmtBuf() *StmtBuf

NewStmtBuf creates a StmtBuf.

func (*StmtBuf) AdvanceOne Uses

func (buf *StmtBuf) AdvanceOne() CmdPos

AdvanceOne advances the cursor one Command over. The command over which the cursor will be positioned when this returns may not be in the buffer yet. The previous CmdPos is returned.

func (*StmtBuf) Close Uses

func (buf *StmtBuf) Close()

Close marks the buffer as closed. Once Close() is called, no further push()es are allowed. If a reader is blocked on a CurCmd() call, it is unblocked with io.EOF. Any further CurCmd() call also returns io.EOF (even if some commands were already available in the buffer before the Close()).

Close() is idempotent.

func (*StmtBuf) CurCmd Uses

func (buf *StmtBuf) CurCmd() (Command, CmdPos, error)

CurCmd returns the Command currently indicated by the cursor. Besides the Command itself, the command's position is also returned; the position can be used to later rewind() to this Command.

If the cursor is positioned over an empty slot, the call blocks until the next Command is pushed into the buffer.

If the buffer has previously been Close()d, or is closed while this is blocked, io.EOF is returned.

func (*StmtBuf) Init Uses

func (buf *StmtBuf) Init()

Init initializes a StmtBuf. It exists to avoid the allocation imposed by NewStmtBuf.

func (*StmtBuf) Push Uses

func (buf *StmtBuf) Push(ctx context.Context, cmd Command) error

Push adds a Command to the end of the buffer. If a CurCmd() call was blocked waiting for this command to arrive, it will be woken up.

An error is returned if the buffer has been closed.

func (*StmtBuf) Rewind Uses

func (buf *StmtBuf) Rewind(ctx context.Context, pos CmdPos)

Rewind resets the buffer's position to pos.

type StmtBufReader Uses

type StmtBufReader struct {
    // contains filtered or unexported fields
}

StmtBufReader is an exported interface for reading a StmtBuf. Normally only the write interface of the buffer is exported, as it is used by the pgwire.

func MakeStmtBufReader Uses

func MakeStmtBufReader(buf *StmtBuf) StmtBufReader

MakeStmtBufReader creates a StmtBufReader.

func (*StmtBufReader) AdvanceOne Uses

func (r *StmtBufReader) AdvanceOne()

AdvanceOne moves the cursor one position over.

func (StmtBufReader) CurCmd Uses

func (r StmtBufReader) CurCmd() (Command, error)

CurCmd returns the current command in the buffer.

func (*StmtBufReader) SeekToNextBatch Uses

func (r *StmtBufReader) SeekToNextBatch() error

SeekToNextBatch skips to the beginning of the next batch of commands.

type Sync Uses

type Sync struct{}

Sync is a command that serves two purposes: 1) It marks the end of one batch of commands and the beginning of the next. stmtBuf.seekToNextBatch will seek to this marker. 2) It generates a ReadyForQuery protocol message.

A Sync command is generated for both the simple and the extended pgwire protocol variants. So, it doesn't strictly correspond to a pgwire sync message - those are not sent in the simple protocol. We synthesize Sync commands though because their handling matches the simple protocol too.

func (Sync) String Uses

func (Sync) String() string

type SyncResult Uses

type SyncResult interface {
    ResultBase
}

SyncResult represents the result of a Sync command. When closed, a readyForQuery message will be generated and all buffered data will be flushed.

type SyncSchemaChangersFilter Uses

type SyncSchemaChangersFilter func(TestingSchemaChangerCollection)

SyncSchemaChangersFilter is the type of a hook to be installed through the ExecutorContext for blocking or otherwise manipulating schema changers run through the sync schema changers path.

type TableCollection Uses

type TableCollection struct {
    // contains filtered or unexported fields
}

TableCollection is a collection of tables held by a single session that serves SQL requests, or a background job using a table descriptor. The collection is cleared using releaseTables() which is called at the end of each transaction on the session, or on hitting conditions such as errors, or retries that result in transaction timestamp changes.

type TableDescriptor Uses

type TableDescriptor = sqlbase.TableDescriptor

TableDescriptor is provided for convenience and to make the interface definitions below more intuitive.

type TableNames Uses

type TableNames = tree.TableNames

TableNames is provided for convenience and to make the interface definitions below more intuitive.

func GetObjectNames Uses

func GetObjectNames(
    ctx context.Context,
    txn *client.Txn,
    sc SchemaResolver,
    dbDesc *DatabaseDescriptor,
    scName string,
    explicitPrefix bool,
) (res TableNames, err error)

GetObjectNames retrieves the names of all objects in the target database/schema.

type TestingSchemaChangerCollection Uses

type TestingSchemaChangerCollection struct {
    // contains filtered or unexported fields
}

TestingSchemaChangerCollection is an exported (for testing) version of schemaChangerCollection. TODO(andrei): get rid of this type once we can have tests internal to the sql package (as of April 2016 we can't because sql can't import server).

func (TestingSchemaChangerCollection) ClearSchemaChangers Uses

func (tscc TestingSchemaChangerCollection) ClearSchemaChangers()

ClearSchemaChangers clears the schema changers from the collection. If this is called from a SyncSchemaChangersFilter, no schema changer will be run.

type TransactionStatusIndicator Uses

type TransactionStatusIndicator byte

TransactionStatusIndicator represents a pg identifier for the transaction state.

const (
    // IdleTxnBlock means the session is outside of a transaction.
    IdleTxnBlock TransactionStatusIndicator = 'I'
    // InTxnBlock means the session is inside a transaction.
    InTxnBlock TransactionStatusIndicator = 'T'
    // InFailedTxnBlock means the session is inside a transaction, but the
    // transaction is in the Aborted state.
    InFailedTxnBlock TransactionStatusIndicator = 'E'
)

type UncachedDatabaseDescriptor Uses

type UncachedDatabaseDescriptor = sqlbase.DatabaseDescriptor

UncachedDatabaseDescriptor is provided for convenience and to make the interface definitions below more intuitive.

type UncachedPhysicalAccessor Uses

type UncachedPhysicalAccessor struct{}

UncachedPhysicalAccessor implements direct access to DB descriptors, without any kind of caching.

func (UncachedPhysicalAccessor) GetDatabaseDesc Uses

func (a UncachedPhysicalAccessor) GetDatabaseDesc(
    ctx context.Context, txn *client.Txn, name string, flags tree.DatabaseLookupFlags,
) (desc *DatabaseDescriptor, err error)

GetDatabaseDesc implements the SchemaAccessor interface.

func (UncachedPhysicalAccessor) GetObjectDesc Uses

func (a UncachedPhysicalAccessor) GetObjectDesc(
    ctx context.Context, txn *client.Txn, name *ObjectName, flags tree.ObjectLookupFlags,
) (ObjectDescriptor, error)

GetObjectDesc implements the SchemaAccessor interface.

func (UncachedPhysicalAccessor) GetObjectNames Uses

func (a UncachedPhysicalAccessor) GetObjectNames(
    ctx context.Context,
    txn *client.Txn,
    dbDesc *DatabaseDescriptor,
    scName string,
    flags tree.DatabaseListFlags,
) (TableNames, error)

GetObjectNames implements the SchemaAccessor interface.

func (UncachedPhysicalAccessor) IsValidSchema Uses

func (a UncachedPhysicalAccessor) IsValidSchema(dbDesc *DatabaseDescriptor, scName string) bool

IsValidSchema implements the SchemaAccessor interface.

type ViewDescriptor Uses

type ViewDescriptor = sqlbase.TableDescriptor

ViewDescriptor is provided for convenience and to make the interface definitions below more intuitive.

type VirtualSchemaHolder Uses

type VirtualSchemaHolder struct {
    // contains filtered or unexported fields
}

VirtualSchemaHolder is a type used to provide convenient access to virtual database and table descriptors. VirtualSchemaHolder, virtualSchemaEntry, and virtualDefEntry make up the generated data structure which the virtualSchemas slice is mapped to. Because of this, they should not be created directly, but instead will be populated in a post-startup hook on an Executor.

func NewVirtualSchemaHolder Uses

func NewVirtualSchemaHolder(
    ctx context.Context, st *cluster.Settings,
) (*VirtualSchemaHolder, error)

NewVirtualSchemaHolder creates a new VirtualSchemaHolder.

type VirtualTabler Uses

type VirtualTabler interface {
    // contains filtered or unexported methods
}

VirtualTabler is used to fetch descriptors for virtual tables and databases.

Directories

PathSynopsis
backfill
colencoding
colexec
colexec/execerror
colexec/execgen
colexec/execpb
colexec/typeconv
colflow
colflow/colrpc
covering
delegate
distsql
execinfra
execinfrapb
flowinfra
lex
logictest
mutations
optPackage opt contains the Cockroach SQL optimizer.
opt/benchPackage bench houses benchmarks for the SQL optimizer.
opt/catPackage cat contains interfaces that are used by the query optimizer to avoid including specifics of sqlbase structures in the opt code.
opt/constraint
opt/exec
opt/exec/execbuilder
opt/idxconstraint
opt/memo
opt/norm
opt/opbench
opt/optbuilder
opt/optgen/exprgen
opt/optgen/langPackage lang implements a language called Optgen, short for "optimizer generator".
opt/orderingPackage ordering contains operator-specific logic related to orderings - whether ops can provide Required orderings, what orderings do they need to require from their children, etc.
opt/props
opt/props/physical
opt/testutils
opt/testutils/opttester
opt/testutils/testcat
opt/testutils/testexpr
opt/xform
parser
parser/fuzz
pgwire
pgwire/hbaPackage hba implements an hba.conf parser.
pgwire/pgcodePackage pgcode defines the PostgreSQL 5-character support codes used throughout the CockroachDB source tree.
pgwire/pgerror
pgwire/pgwirebasePackage pgwirebase contains type definitions and very basic protocol structures to be used by both the pgwire package and by others (particularly by the sql package).
physicalplan
physicalplan/replicaoraclePackage replicaoracle provides functionality for physicalplan to choose a replica for a range.
privilege
querycache
row
rowcontainer
rowexec
rowflow
schemachange
scrub
sem/builtins
sem/transform
sem/tree
sessiondata
sqlbase
sqltelemetryPackage sqltelemetry contains telemetry counter definitions for various SQL features.
sqlutil
stats
tests
types
vtable

Package sql imports 132 packages (graph) and is imported by 89 packages. Updated 2019-11-20. Refresh now. Tools for package owners.