Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 3 additions & 1 deletion .github/workflows/integration-cassandra.yml
Original file line number Diff line number Diff line change
Expand Up @@ -145,8 +145,10 @@ jobs:
run: pip install -r requirements.txt
- name: Build release (cassandra backend)
run: cargo build --release --no-default-features --features cassandra
- name: Install Terraform (pinned version, checksum asserted)
run: echo "TERRAFORM=$(tests/terraform/install-terraform.sh "$RUNNER_TEMP/terraform")" >> "$GITHUB_ENV"
- name: Run Cassandra pytest suite
run: devtools/run-cassandra-tests -- --pytest --comprehensive --parallel
run: devtools/run-cassandra-tests -- --pytest --comprehensive --terraform --parallel

cassandra-rust:
runs-on: extenddb_ubuntu-2404_4-core
Expand Down
4 changes: 3 additions & 1 deletion .github/workflows/integration-mongodb.yml
Original file line number Diff line number Diff line change
Expand Up @@ -48,8 +48,10 @@ jobs:
run: pip install -r requirements.txt
- name: Build release (mongodb backend)
run: cargo build --release --no-default-features --features mongodb-test-hooks
- name: Install Terraform (pinned version, checksum asserted)
run: echo "TERRAFORM=$(tests/terraform/install-terraform.sh "$RUNNER_TEMP/terraform")" >> "$GITHUB_ENV"
- name: Run MongoDB pytest suite
run: devtools/run-mongodb-tests -- --pytest --comprehensive --parallel
run: devtools/run-mongodb-tests -- --pytest --comprehensive --terraform --parallel

mongodb-rust:
runs-on: extenddb_ubuntu-2404_4-core
Expand Down
10 changes: 8 additions & 2 deletions .github/workflows/integration.yml
Original file line number Diff line number Diff line change
Expand Up @@ -89,13 +89,16 @@ jobs:
- name: Install Python dependencies
run: pip install -r requirements.txt

- name: Install Terraform (pinned version, checksum asserted)
run: echo "TERRAFORM=$(tests/terraform/install-terraform.sh "$RUNNER_TEMP/terraform")" >> "$GITHUB_ENV"

- name: Run integration tests
env:
EXTENDDB_TEST_ENDPOINT: https://127.0.0.1:18443
EXTENDDB_ADMIN_USER: admin
EXTENDDB_ADMIN_PASSWORD: ${{ steps.init.outputs.admin_password }}
EXTENDDB_TEST_PG_ADMIN_CONNECTION_STRING: postgresql://postgres:devpass@127.0.0.1:5432
run: devtools/run-tests --extenddb --pytest --comprehensive --parallel --filter "not import_export"
run: devtools/run-tests --extenddb --pytest --comprehensive --terraform --parallel --filter "not import_export"

# run-tests restarts the server daemonized to apply the import/export
# config, and a daemonized server logs to syslog, not the job log. Without
Expand Down Expand Up @@ -150,12 +153,15 @@ jobs:
- name: Install Python dependencies
run: pip install -r requirements.txt

- name: Install Terraform (pinned version, checksum asserted)
run: echo "TERRAFORM=$(tests/terraform/install-terraform.sh "$RUNNER_TEMP/terraform")" >> "$GITHUB_ENV"

- name: Run integration tests
env:
EXTENDDB_TEST_ENDPOINT: https://127.0.0.1:18443
EXTENDDB_ADMIN_USER: admin
EXTENDDB_ADMIN_PASSWORD: ${{ steps.init.outputs.admin_password }}
run: devtools/run-tests --extenddb --pytest --comprehensive --parallel --filter "not import_export"
run: devtools/run-tests --extenddb --pytest --comprehensive --terraform --parallel --filter "not import_export"

# Same rationale as the postgres job: the daemonized server logs to
# syslog, so this dump is the only server-side evidence on failure.
Expand Down
5 changes: 5 additions & 0 deletions crates/core/src/types/mod.rs
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,7 @@ mod query;
mod stream;
mod table;
mod transaction;
mod warm_throughput;

pub use attribute_value::AttributeValue;
pub use backup::{
Expand Down Expand Up @@ -76,3 +77,7 @@ pub use transaction::{
TransactGetItem, TransactGetItemsInput, TransactGetItemsOutput, TransactPut, TransactUpdate,
TransactWriteItem, TransactWriteItemsInput, TransactWriteItemsOutput,
};
pub use warm_throughput::{
ON_DEMAND_WARM_READ_UNITS, ON_DEMAND_WARM_WRITE_UNITS, WarmThroughput, WarmThroughputDefaults,
WarmThroughputDescription, WarmThroughputRecord, WarmThroughputStatus, epoch_seconds_now,
};
82 changes: 79 additions & 3 deletions crates/core/src/types/table.rs
Original file line number Diff line number Diff line change
Expand Up @@ -303,6 +303,14 @@ pub struct GsiDescription {
pub item_count: i64,
#[serde(rename = "IndexArn")]
pub index_arn: String,
/// Derived by `TableDescription::prepare_for_response` from
/// `warm_throughput_record`; backends leave this `None`.
#[serde(rename = "WarmThroughput", skip_serializing_if = "Option::is_none")]
pub warm_throughput: Option<super::WarmThroughputDescription>,
/// What the backend stored for this index, if a client ever set it. Not a
/// wire member; see [`super::WarmThroughputRecord`]. Nothing acts on it.
#[serde(skip)]
pub warm_throughput_record: Option<super::WarmThroughputRecord>,
}

/// Local secondary index description returned in responses.
Expand Down Expand Up @@ -336,6 +344,12 @@ pub struct GsiInput {
skip_serializing_if = "Option::is_none"
)]
pub provisioned_throughput: Option<ProvisionedThroughput>,
#[serde(
rename = "WarmThroughput",
default,
skip_serializing_if = "Option::is_none"
)]
pub warm_throughput: Option<super::WarmThroughput>,
}

/// Local secondary index definition for `CreateTable` requests.
Expand Down Expand Up @@ -822,6 +836,46 @@ impl TableDescription {
} else if let Some(summary) = self.restore_summary.as_mut() {
summary.restore_in_progress = true;
}
self.populate_warm_throughput(super::epoch_seconds_now());
}

/// Derive the `WarmThroughput` member of the table and of each global
/// secondary index from the stored records (see
/// [`super::WarmThroughputRecord::describe_for`] for the measured rules).
/// Local secondary indexes carry none. Reached through
/// [`Self::prepare_for_response`]; `now` is epoch seconds.
pub fn populate_warm_throughput(&mut self, now: f64) {
let pay_per_request = self
.billing_mode_summary
.as_ref()
.is_some_and(|s| s.billing_mode == BillingMode::PayPerRequest);
let table_defaults = super::WarmThroughputDefaults::for_billing(
pay_per_request,
self.provisioned_throughput.read_capacity_units,
self.provisioned_throughput.write_capacity_units,
);
self.warm_throughput = super::WarmThroughputRecord::describe_for(
self.warm_throughput_record.as_ref(),
self.table_status == TableStatus::Creating,
table_defaults,
now,
);
for gsi in self.global_secondary_indexes.iter_mut().flatten() {
let (rcu, wcu) = gsi.provisioned_throughput.as_ref().map_or((0, 0), |pt| {
(pt.read_capacity_units, pt.write_capacity_units)
});
let defaults = super::WarmThroughputDefaults::for_billing(pay_per_request, rcu, wcu);
// An index created with its table is CREATING while the table is;
// one added by UpdateTable is CREATING while it backfills.
let creating =
self.table_status == TableStatus::Creating || gsi.index_status == "CREATING";
gsi.warm_throughput = super::WarmThroughputRecord::describe_for(
gsi.warm_throughput_record.as_ref(),
creating,
defaults,
now,
);
}
}

/// Populate `TableThroughputModeSummary` as an exact mirror of
Expand Down Expand Up @@ -901,6 +955,14 @@ pub struct TableDescription {
pub on_demand_throughput: Option<OnDemandThroughput>,
#[serde(rename = "RestoreSummary", skip_serializing_if = "Option::is_none")]
pub restore_summary: Option<RestoreSummary>,
/// Derived by [`Self::prepare_for_response`] from `warm_throughput_record`,
/// the billing mode and the provisioned units; backends leave this `None`.
#[serde(rename = "WarmThroughput", skip_serializing_if = "Option::is_none")]
pub warm_throughput: Option<super::WarmThroughputDescription>,
/// What the backend stored for the table, if a client ever set it. Not a
/// wire member; see [`super::WarmThroughputRecord`]. Nothing acts on it.
#[serde(skip)]
pub warm_throughput_record: Option<super::WarmThroughputRecord>,
}

/// Details of a restore reported on the restored table's description.
Expand Down Expand Up @@ -963,6 +1025,8 @@ pub struct CreateTableInput {
pub table_class: Option<String>,
#[serde(rename = "OnDemandThroughput")]
pub on_demand_throughput: Option<OnDemandThroughput>,
#[serde(rename = "WarmThroughput", default)]
pub warm_throughput: Option<super::WarmThroughput>,
}

/// `CreateTable` response body.
Expand Down Expand Up @@ -1063,6 +1127,8 @@ pub struct CreateGsiAction {
skip_serializing_if = "Option::is_none"
)]
pub provisioned_throughput: Option<ProvisionedThroughput>,
#[serde(rename = "WarmThroughput", default)]
pub warm_throughput: Option<super::WarmThroughput>,
}

/// Delete an existing GSI from a table.
Expand All @@ -1072,14 +1138,17 @@ pub struct DeleteGsiAction {
pub index_name: String,
}

/// Update provisioned throughput on an existing GSI.
/// Update an existing GSI.
///
/// Recognized by the deserializer but not yet implemented — the engine
/// returns a clear "not yet supported" error.
/// `WarmThroughput` is stored and reported (see [`super::WarmThroughputRecord`]).
/// A provisioned-throughput update is recognized by the deserializer but not
/// yet implemented: the engine returns a clear "not yet supported" error.
#[derive(Debug, Clone, PartialEq, Deserialize)]
pub struct UpdateGsiAction {
#[serde(rename = "IndexName")]
pub index_name: String,
#[serde(rename = "WarmThroughput", default)]
pub warm_throughput: Option<super::WarmThroughput>,
}

/// `UpdateTable` request body.
Expand Down Expand Up @@ -1110,6 +1179,13 @@ pub struct UpdateTableInput {
/// unaffected.
#[serde(rename = "VectorIndexUpdates", default)]
pub vector_index_updates: Option<Vec<VectorIndexUpdate>>,
#[serde(rename = "WarmThroughput", default)]
pub warm_throughput: Option<super::WarmThroughput>,
/// Accepted so that a request carrying it is recognised: `UpdateTable`
/// does not act on it (as before), but a table-level `WarmThroughput`
/// beside it is refused with the service's sentence.
#[serde(rename = "SSESpecification", default)]
pub sse_specification: Option<serde_json::Value>,
}

/// `UpdateTable` response body.
Expand Down
Loading
Loading