You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kudu.apache.org by al...@apache.org on 2022/08/11 02:05:47 UTC
[kudu] branch master updated: KUDU-3353 [schema] Add an immutable attribute on column schema (part 1)
This is an automated email from the ASF dual-hosted git repository.
alexey pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/kudu.git
The following commit(s) were added to refs/heads/master by this push:
new b6eedb224 KUDU-3353 [schema] Add an immutable attribute on column schema (part 1)
b6eedb224 is described below
commit b6eedb224f715ad86378a92d25f09c2084b0e2b7
Author: Yingchun Lai <la...@apache.org>
AuthorDate: Sun Jul 17 23:48:26 2022 +0800
KUDU-3353 [schema] Add an immutable attribute on column schema (part 1)
The overview of design:
1. Add a new column attribute IMMUTABLE, meaning the column cell
value can not be updated after it's been written during inserting
the row.
2. An attempt to modify an immutable cell of an existing row by
UPDATE or UPSERT operation results in returning the newly added
Status::IsImmutable().
3. Use UPDATE_IGNORE and add UPSERT_IGNORE, for UPDATE and UPSERT
ops but ignore update errors on IMMUTABLE columns. Note that
the rest of the columns are updated accordingly to the
operation's data, only the immutable columns aren't changed.
With this change, UPDATE_IGNORE ops ignore both 'key not found'
and 'update on immutable column' errors.
Some use cases:
1. A column represents a semantically constant entity. The
corresponding value is present in every row for a particular
primary key and might change, but it's captured upon the very
first occurrence. An example is 'first_login_timestamp' for
a particular user while 'login_timestamp' is present in every
login record.
2. Similar to the item 1, but the corresponding value, if present,
is the same with every record for a particular primary key.
Here the intention is to reduce the length of the column's
change list. An example is a 'birthday' column.
This patch includes the changes on the server side, proto files, and
necessary changes on the client side because of ColumnSchema
constructor's been changed.
Change-Id: I01e5a806c0e873239b49e6d0b37a7e36578b508d
Reviewed-on: http://gerrit.cloudera.org:8080/18742
Tested-by: Kudu Jenkins
Reviewed-by: Alexey Serbin <al...@apache.org>
---
.../org/apache/kudu/client/AsyncKuduClient.java | 2 +
.../org/apache/kudu/client/TestKuduSession.java | 1 +
src/kudu/client/client-internal.cc | 4 +-
src/kudu/client/client-test.cc | 1 +
src/kudu/client/scan_configuration.cc | 1 +
src/kudu/client/schema.cc | 1 +
src/kudu/codegen/codegen-test.cc | 10 ++-
src/kudu/common/column_predicate-test.cc | 3 +-
src/kudu/common/common.proto | 2 +
src/kudu/common/generic_iterators-test.cc | 1 +
src/kudu/common/partial_row-test.cc | 4 +-
src/kudu/common/partition-test.cc | 4 +-
src/kudu/common/row_operations-test.cc | 8 +-
src/kudu/common/row_operations.cc | 16 ++++
src/kudu/common/row_operations.h | 7 +-
src/kudu/common/row_operations.proto | 1 +
src/kudu/common/schema-test.cc | 23 +++---
src/kudu/common/schema.cc | 18 ++++-
src/kudu/common/schema.h | 28 +++++--
src/kudu/common/wire_protocol-test.cc | 8 +-
src/kudu/common/wire_protocol.cc | 9 ++-
src/kudu/common/wire_protocol.proto | 1 +
src/kudu/consensus/log-test.cc | 5 +-
src/kudu/master/master.proto | 2 +
src/kudu/master/master_service.cc | 11 ++-
src/kudu/tablet/all_types-scan-correctness-test.cc | 10 +--
src/kudu/tablet/cfile_set-test.cc | 4 +-
src/kudu/tablet/diskrowset-test.cc | 12 ++-
src/kudu/tablet/local_tablet_writer.h | 4 +
src/kudu/tablet/ops/op.cc | 2 +
src/kudu/tablet/ops/op.h | 1 +
src/kudu/tablet/ops/write_op.cc | 10 +++
src/kudu/tablet/ops/write_op.h | 2 +-
src/kudu/tablet/row_op.cc | 1 +
src/kudu/tablet/tablet-decoder-eval-test.cc | 6 +-
src/kudu/tablet/tablet-test-base.h | 63 ++++++++++++++-
src/kudu/tablet/tablet-test-util.h | 2 +-
src/kudu/tablet/tablet-test.cc | 93 +++++++++++++++++++++-
src/kudu/tablet/tablet.cc | 52 +++++++++++-
src/kudu/tablet/tablet.h | 2 +-
src/kudu/tablet/tablet_bootstrap.cc | 6 +-
src/kudu/tablet/tablet_metrics.cc | 10 +++
src/kudu/tablet/tablet_metrics.h | 1 +
src/kudu/tablet/tablet_random_access-test.cc | 18 ++++-
src/kudu/tablet/txn_participant-test.cc | 2 +
src/kudu/tools/kudu-tool-test.cc | 13 ++-
.../tserver/tablet_server_authorization-test.cc | 9 ++-
src/kudu/tserver/tserver.proto | 2 +
src/kudu/util/status.cc | 3 +
src/kudu/util/status.h | 10 +++
50 files changed, 429 insertions(+), 80 deletions(-)
diff --git a/java/kudu-client/src/main/java/org/apache/kudu/client/AsyncKuduClient.java b/java/kudu-client/src/main/java/org/apache/kudu/client/AsyncKuduClient.java
index 1b1a690ca..303c641f9 100644
--- a/java/kudu-client/src/main/java/org/apache/kudu/client/AsyncKuduClient.java
+++ b/java/kudu-client/src/main/java/org/apache/kudu/client/AsyncKuduClient.java
@@ -2717,6 +2717,8 @@ public class AsyncKuduClient implements AutoCloseable {
return AsyncUtil.addBoth(response, new PingSupportsFeatureCallback());
}
+ // TODO(yingchun): also need add 'public Deferred<Boolean> supportsUpsertIgnoreOperations()'
+
private static final class PingSupportsFeatureCallback implements Callback<Boolean, Object> {
@Override
public Boolean call(final Object resp) {
diff --git a/java/kudu-client/src/test/java/org/apache/kudu/client/TestKuduSession.java b/java/kudu-client/src/test/java/org/apache/kudu/client/TestKuduSession.java
index 905ccbe74..422c2e6bb 100644
--- a/java/kudu-client/src/test/java/org/apache/kudu/client/TestKuduSession.java
+++ b/java/kudu-client/src/test/java/org/apache/kudu/client/TestKuduSession.java
@@ -356,6 +356,7 @@ public class TestKuduSession {
assertEquals(successfulInserts, metrics.getMetric("successful_inserts"));
assertEquals(insertIgnoreErrors, metrics.getMetric("insert_ignore_errors"));
assertEquals(successfulUpserts, metrics.getMetric("successful_upserts"));
+ // TODO(yingchun): should test upsert_ignore_errors
assertEquals(successfulUpdates, metrics.getMetric("successful_updates"));
assertEquals(updateIgnoreErrors, metrics.getMetric("update_ignore_errors"));
assertEquals(successfulDeletes, metrics.getMetric("successful_deletes"));
diff --git a/src/kudu/client/client-internal.cc b/src/kudu/client/client-internal.cc
index 9095ae9c2..0c8a6f7bb 100644
--- a/src/kudu/client/client-internal.cc
+++ b/src/kudu/client/client-internal.cc
@@ -27,6 +27,7 @@
#include <ostream>
#include <random>
#include <string>
+#include <type_traits>
#include <utility>
#include <vector>
@@ -55,9 +56,7 @@
#include "kudu/master/master.proxy.h"
#include "kudu/master/txn_manager.proxy.h"
#include "kudu/rpc/connection.h"
-#include "kudu/rpc/messenger.h"
#include "kudu/rpc/request_tracker.h"
-#include "kudu/rpc/response_callback.h"
#include "kudu/rpc/rpc.h"
#include "kudu/rpc/rpc_controller.h"
#include "kudu/security/cert.h"
@@ -312,6 +311,7 @@ Status KuduClient::Data::GetTabletServer(KuduClient* client,
return Status::OK();
}
+// TODO(yingchun): Add has_immutable_column_schema
Status KuduClient::Data::CreateTable(KuduClient* client,
const CreateTableRequestPB& req,
CreateTableResponsePB* resp,
diff --git a/src/kudu/client/client-test.cc b/src/kudu/client/client-test.cc
index c86288e13..9946afd1f 100644
--- a/src/kudu/client/client-test.cc
+++ b/src/kudu/client/client-test.cc
@@ -2948,6 +2948,7 @@ static void DoVerifyMetrics(const KuduSession* session,
ASSERT_EQ(successful_inserts, metrics["successful_inserts"]);
ASSERT_EQ(insert_ignore_errors, metrics["insert_ignore_errors"]);
ASSERT_EQ(successful_upserts, metrics["successful_upserts"]);
+ // TODO(yingchun): should test upsert_ignore_errors
ASSERT_EQ(successful_updates, metrics["successful_updates"]);
ASSERT_EQ(update_ignore_errors, metrics["update_ignore_errors"]);
ASSERT_EQ(successful_deletes, metrics["successful_deletes"]);
diff --git a/src/kudu/client/scan_configuration.cc b/src/kudu/client/scan_configuration.cc
index 707efc297..1d5968a2a 100644
--- a/src/kudu/client/scan_configuration.cc
+++ b/src/kudu/client/scan_configuration.cc
@@ -242,6 +242,7 @@ Status ScanConfiguration::AddIsDeletedColumn() {
ColumnSchema is_deleted(col_name,
IS_DELETED,
/*is_nullable=*/false,
+ /*is_immutable=*/false,
&read_default);
cols.emplace_back(std::move(is_deleted));
diff --git a/src/kudu/client/schema.cc b/src/kudu/client/schema.cc
index b14d1f5be..e3e48324f 100644
--- a/src/kudu/client/schema.cc
+++ b/src/kudu/client/schema.cc
@@ -746,6 +746,7 @@ KuduColumnSchema::KuduColumnSchema(const string &name,
type_attr_private.length = type_attributes.length();
col_ = new ColumnSchema(name, ToInternalDataType(type, type_attributes),
is_nullable,
+ false, // TODO(yingchun): set according to a new added parameter later
default_value, default_value, attr_private,
type_attr_private, comment);
}
diff --git a/src/kudu/codegen/codegen-test.cc b/src/kudu/codegen/codegen-test.cc
index b8ccead0d..0dca7035f 100644
--- a/src/kudu/codegen/codegen-test.cc
+++ b/src/kudu/codegen/codegen-test.cc
@@ -23,10 +23,12 @@
#include <string>
#include <vector>
+// IWYU pragma: no_include "testing/base/public/gunit.h"
#include <gflags/gflags_declare.h>
#include <glog/logging.h>
#include <glog/stl_logging.h> // IWYU pragma: keep
#include <gmock/gmock-matchers.h>
+#include <gtest/gtest-matchers.h>
#include <gtest/gtest.h>
#include "kudu/codegen/code_generator.h"
@@ -80,10 +82,10 @@ class CodegenTest : public KuduTest {
base_ = SchemaBuilder(base_).Build(); // add IDs
// Create an extended default schema
- cols.emplace_back("int32-R ", INT32, false, kI32R, nullptr);
- cols.emplace_back("int32-RW", INT32, false, kI32R, kI32W);
- cols.emplace_back("str32-R ", STRING, false, kStrR, nullptr);
- cols.emplace_back("str32-RW", STRING, false, kStrR, kStrW);
+ cols.emplace_back("int32-R ", INT32, false, false, kI32R, nullptr);
+ cols.emplace_back("int32-RW", INT32, false, false, kI32R, kI32W);
+ cols.emplace_back("str32-R ", STRING, false, false, kStrR, nullptr);
+ cols.emplace_back("str32-RW", STRING, false, false, kStrR, kStrW);
defaults_.Reset(cols, 1);
defaults_ = SchemaBuilder(defaults_).Build(); // add IDs
diff --git a/src/kudu/common/column_predicate-test.cc b/src/kudu/common/column_predicate-test.cc
index 39f584112..18de259a8 100644
--- a/src/kudu/common/column_predicate-test.cc
+++ b/src/kudu/common/column_predicate-test.cc
@@ -1462,7 +1462,8 @@ TEST_F(TestColumnPredicate, TestEquals) {
ASSERT_NE(ColumnPredicate::None(c1), ColumnPredicate::None(c1string));
const int kDefaultOf3 = 3;
- ColumnSchema c1dflt("c1", INT32, /*is_nullable=*/false, /*read_default=*/&kDefaultOf3);
+ ColumnSchema c1dflt("c1", INT32, /*is_nullable=*/false,
+ /*is_immutable=*/false, /*read_default=*/&kDefaultOf3);
ASSERT_NE(ColumnPredicate::None(c1), ColumnPredicate::None(c1dflt));
}
diff --git a/src/kudu/common/common.proto b/src/kudu/common/common.proto
index cf41ec306..2d43328c0 100644
--- a/src/kudu/common/common.proto
+++ b/src/kudu/common/common.proto
@@ -137,6 +137,8 @@ message ColumnSchemaPB {
// The comment for the column.
optional string comment = 12;
+
+ optional bool immutable = 13 [default = false];
}
message ColumnSchemaDeltaPB {
diff --git a/src/kudu/common/generic_iterators-test.cc b/src/kudu/common/generic_iterators-test.cc
index 647443eb6..4cc1f4b67 100644
--- a/src/kudu/common/generic_iterators-test.cc
+++ b/src/kudu/common/generic_iterators-test.cc
@@ -87,6 +87,7 @@ static const bool kIsDeletedReadDefault = false;
static const Schema kIntSchemaWithVCol({ ColumnSchema("val", INT64),
ColumnSchema("is_deleted", IS_DELETED,
/*is_nullable=*/false,
+ /*is_immutable=*/false,
/*read_default=*/&kIsDeletedReadDefault) },
/*key_columns=*/1);
diff --git a/src/kudu/common/partial_row-test.cc b/src/kudu/common/partial_row-test.cc
index a62f30919..5aaa817ce 100644
--- a/src/kudu/common/partial_row-test.cc
+++ b/src/kudu/common/partial_row-test.cc
@@ -42,9 +42,9 @@ class PartialRowTest : public KuduTest {
ColumnSchema("int_val", INT32),
ColumnSchema("string_val", STRING, true),
ColumnSchema("binary_val", BINARY, true),
- ColumnSchema("decimal_val", DECIMAL32, true, nullptr, nullptr,
+ ColumnSchema("decimal_val", DECIMAL32, true, false, nullptr, nullptr,
ColumnStorageAttributes(), ColumnTypeAttributes(6, 2)),
- ColumnSchema("varchar_val", VARCHAR, true, nullptr, nullptr,
+ ColumnSchema("varchar_val", VARCHAR, true, false, nullptr, nullptr,
ColumnStorageAttributes(), ColumnTypeAttributes(10)) },
1) {
SeedRandom();
diff --git a/src/kudu/common/partition-test.cc b/src/kudu/common/partition-test.cc
index a230bdc37..de40eaea0 100644
--- a/src/kudu/common/partition-test.cc
+++ b/src/kudu/common/partition-test.cc
@@ -876,9 +876,9 @@ TEST_F(PartitionTest, TestIncrementRangePartitionStringBounds) {
}
TEST_F(PartitionTest, TestVarcharRangePartitions) {
- Schema schema({ ColumnSchema("c1", VARCHAR, false, nullptr, nullptr,
+ Schema schema({ ColumnSchema("c1", VARCHAR, false, false, nullptr, nullptr,
ColumnStorageAttributes(), ColumnTypeAttributes(10)),
- ColumnSchema("c2", VARCHAR, false, nullptr, nullptr,
+ ColumnSchema("c2", VARCHAR, false, false, nullptr, nullptr,
ColumnStorageAttributes(), ColumnTypeAttributes(10)) },
{ ColumnId(0), ColumnId(1) }, 2);
diff --git a/src/kudu/common/row_operations-test.cc b/src/kudu/common/row_operations-test.cc
index 271220ea5..17c336dda 100644
--- a/src/kudu/common/row_operations-test.cc
+++ b/src/kudu/common/row_operations-test.cc
@@ -126,7 +126,7 @@ void RowOperationsTest::CheckDecodeDoesntCrash(const Schema& client_schema,
void RowOperationsTest::DoFuzzTest(const Schema& server_schema,
const KuduPartialRow& row,
int n_random_changes) {
- for (int operation = 0; operation <= 11; operation++) {
+ for (int operation = 0; operation <= 12; operation++) {
RowOperationsPB pb;
RowOperationsPBEncoder enc(&pb);
@@ -166,6 +166,8 @@ void RowOperationsTest::DoFuzzTest(const Schema& server_schema,
break;
case 11:
enc.Add(RowOperationsPB::DELETE_IGNORE, row);
+ case 12:
+ enc.Add(RowOperationsPB::UPSERT_IGNORE, row);
break;
}
@@ -870,6 +872,7 @@ void CheckExceedCellLimit(const Schema& client_schema,
case RowOperationsPB::INSERT_IGNORE:
case RowOperationsPB::UPDATE_IGNORE:
case RowOperationsPB::DELETE_IGNORE:
+ case RowOperationsPB::UPSERT_IGNORE:
s = decoder.DecodeOperations<WRITE_OPS>(&ops);
break;
case RowOperationsPB::SPLIT_ROW:
@@ -896,7 +899,8 @@ void CheckInsertUpsertExceedCellLimit(const Schema& client_schema,
const string& expect_msg) {
for (auto op_type : { RowOperationsPB::INSERT,
RowOperationsPB::INSERT_IGNORE,
- RowOperationsPB::UPSERT }) {
+ RowOperationsPB::UPSERT,
+ RowOperationsPB::UPSERT_IGNORE }) {
NO_FATALS(CheckExceedCellLimit(client_schema, col_values, op_type, expect_status, expect_msg));
}
}
diff --git a/src/kudu/common/row_operations.cc b/src/kudu/common/row_operations.cc
index 84cfbce59..20ac21444 100644
--- a/src/kudu/common/row_operations.cc
+++ b/src/kudu/common/row_operations.cc
@@ -20,6 +20,7 @@
#include <cstring>
#include <ostream>
#include <string>
+#include <type_traits>
#include <utility>
#include <gflags/gflags.h>
@@ -69,6 +70,8 @@ string DecodedRowOperation::ToString(const Schema& schema) const {
return "INSERT IGNORE " + schema.DebugRow(ConstContiguousRow(&schema, row_data));
case RowOperationsPB::UPSERT:
return "UPSERT " + schema.DebugRow(ConstContiguousRow(&schema, row_data));
+ case RowOperationsPB::UPSERT_IGNORE:
+ return "UPSERT IGNORE " + schema.DebugRow(ConstContiguousRow(&schema, row_data));
case RowOperationsPB::UPDATE:
case RowOperationsPB::UPDATE_IGNORE:
case RowOperationsPB::DELETE:
@@ -564,6 +567,18 @@ Status RowOperationsPBDecoder::DecodeUpdateOrDelete(const ClientServerMapping& m
const ColumnSchema& col = tablet_schema_->column(tablet_col_idx);
if (BitmapTest(client_isset_map, client_col_idx)) {
+ if (col.is_immutable()) {
+ if (op->type == RowOperationsPB::UPDATE) {
+ op->SetFailureStatusOnce(
+ Status::Immutable("UPDATE not allowed for immutable column", col.ToString()));
+ } else {
+ DCHECK_EQ(RowOperationsPB::UPDATE_IGNORE, op->type);
+ op->error_ignored = true;
+ }
+ RETURN_NOT_OK(ReadColumnAndDiscard(col));
+ // Use 'continue' not 'break' to consume the rest row data.
+ continue;
+ }
bool client_set_to_null = client_schema_->has_nullables() &&
BitmapTest(client_null_map, client_col_idx);
uint8_t scratch[kLargestTypeSize];
@@ -711,6 +726,7 @@ Status RowOperationsPBDecoder::DecodeOp<DecoderMode::WRITE_OPS>(
case RowOperationsPB::INSERT:
case RowOperationsPB::INSERT_IGNORE:
case RowOperationsPB::UPSERT:
+ case RowOperationsPB::UPSERT_IGNORE:
RETURN_NOT_OK(DecodeInsertOrUpsert(prototype_row_storage, mapping, op));
break;
case RowOperationsPB::UPDATE:
diff --git a/src/kudu/common/row_operations.h b/src/kudu/common/row_operations.h
index 657a87596..b69f3f7cb 100644
--- a/src/kudu/common/row_operations.h
+++ b/src/kudu/common/row_operations.h
@@ -75,7 +75,7 @@ class RowOperationsPBEncoder {
struct DecodedRowOperation {
RowOperationsPB::Type type;
- // For INSERT, INSERT_IGNORE, or UPSERT, the whole projected row.
+ // For INSERT, INSERT_IGNORE, UPSERT, or UPSERT_IGNORE, the whole projected row.
// For UPDATE, UPDATE_IGNORE, DELETE, or DELETE_IGNORE, the row key.
const uint8_t* row_data;
@@ -93,6 +93,11 @@ struct DecodedRowOperation {
// Per-row result status.
Status result;
+ // True if an ignore op was ignored due to an error.
+ // As of now, the error could be one of the following:
+ // - UPDATE_IGNORE op on a row to update an immutable column.
+ bool error_ignored = false;
+
// Stringifies, including redaction when appropriate.
std::string ToString(const Schema& schema) const;
diff --git a/src/kudu/common/row_operations.proto b/src/kudu/common/row_operations.proto
index c8c8431fe..d5dea532b 100644
--- a/src/kudu/common/row_operations.proto
+++ b/src/kudu/common/row_operations.proto
@@ -40,6 +40,7 @@ message RowOperationsPB {
INSERT_IGNORE = 10;
UPDATE_IGNORE = 11;
DELETE_IGNORE = 12;
+ UPSERT_IGNORE = 13;
// Used when specifying split rows on table creation.
SPLIT_ROW = 4;
diff --git a/src/kudu/common/schema-test.cc b/src/kudu/common/schema-test.cc
index 6a78cbd46..f5a5b8bb1 100644
--- a/src/kudu/common/schema-test.cc
+++ b/src/kudu/common/schema-test.cc
@@ -232,13 +232,13 @@ TEST_P(ParameterizedSchemaTest, TestCopyAndMove) {
// Test basic functionality of Schema definition with decimal columns
TEST_F(TestSchema, TestSchemaWithDecimal) {
ColumnSchema col1("key", STRING);
- ColumnSchema col2("decimal32val", DECIMAL32, false,
+ ColumnSchema col2("decimal32val", DECIMAL32, false, false,
nullptr, nullptr, ColumnStorageAttributes(),
ColumnTypeAttributes(9, 4));
- ColumnSchema col3("decimal64val", DECIMAL64, true,
+ ColumnSchema col3("decimal64val", DECIMAL64, true, false,
nullptr, nullptr, ColumnStorageAttributes(),
ColumnTypeAttributes(18, 10));
- ColumnSchema col4("decimal128val", DECIMAL128, true,
+ ColumnSchema col4("decimal128val", DECIMAL128, true, false,
nullptr, nullptr, ColumnStorageAttributes(),
ColumnTypeAttributes(38, 2));
@@ -266,16 +266,16 @@ TEST_F(TestSchema, TestSchemaWithDecimal) {
// Test Schema::Equals respects decimal column attributes
TEST_F(TestSchema, TestSchemaEqualsWithDecimal) {
ColumnSchema col1("key", STRING);
- ColumnSchema col_18_10("decimal64val", DECIMAL64, true,
+ ColumnSchema col_18_10("decimal64val", DECIMAL64, true, false,
nullptr, nullptr, ColumnStorageAttributes(),
ColumnTypeAttributes(18, 10));
- ColumnSchema col_18_9("decimal64val", DECIMAL64, true,
+ ColumnSchema col_18_9("decimal64val", DECIMAL64, true, false,
nullptr, nullptr, ColumnStorageAttributes(),
ColumnTypeAttributes(18, 9));
- ColumnSchema col_17_10("decimal64val", DECIMAL64, true,
+ ColumnSchema col_17_10("decimal64val", DECIMAL64, true, false,
nullptr, nullptr, ColumnStorageAttributes(),
ColumnTypeAttributes(17, 10));
- ColumnSchema col_17_9("decimal64val", DECIMAL64, true,
+ ColumnSchema col_17_9("decimal64val", DECIMAL64, true, false,
nullptr, nullptr, ColumnStorageAttributes(),
ColumnTypeAttributes(17, 9));
@@ -295,7 +295,7 @@ TEST_F(TestSchema, TestColumnSchemaEquals) {
ColumnSchema col1("key", STRING);
ColumnSchema col2("key1", STRING);
ColumnSchema col3("key", STRING, true);
- ColumnSchema col4("key", STRING, true, &default_str, &default_str);
+ ColumnSchema col4("key", STRING, true, false, &default_str, &default_str);
ASSERT_TRUE(col1.Equals(col1));
ASSERT_FALSE(col1.Equals(col2, ColumnSchema::COMPARE_NAME));
@@ -422,7 +422,7 @@ TEST_F(TestSchema, TestProjectMissingColumn) {
Schema schema3({ ColumnSchema("val", UINT32), ColumnSchema("non_present", UINT32, true) }, 0);
uint32_t default_value = 15;
Schema schema4({ ColumnSchema("val", UINT32),
- ColumnSchema("non_present", UINT32, false, &default_value) },
+ ColumnSchema("non_present", UINT32, false, false, &default_value) },
0);
RowProjector row_projector(&schema1, &schema2);
@@ -492,7 +492,8 @@ TEST_F(TestSchema, TestGetMappedReadProjection) {
const bool kReadDefault = false;
Schema projection({ ColumnSchema("key", STRING),
ColumnSchema("deleted", IS_DELETED,
- /*is_nullable=*/false, /*read_default=*/&kReadDefault) },
+ /*is_nullable=*/false, /*is_immutable=*/false,
+ /*read_default=*/&kReadDefault) },
1);
Schema mapped;
@@ -520,6 +521,7 @@ TEST_F(TestSchema, TestGetMappedReadProjection) {
Status s = nullable_projection.Reset({ ColumnSchema("key", STRING),
ColumnSchema("deleted", IS_DELETED,
/*is_nullable=*/true,
+ /*is_immutable=*/false,
/*read_default=*/&kReadDefault) },
1);
ASSERT_FALSE(s.ok());
@@ -529,6 +531,7 @@ TEST_F(TestSchema, TestGetMappedReadProjection) {
s = no_default_projection.Reset({ ColumnSchema("key", STRING),
ColumnSchema("deleted", IS_DELETED,
/*is_nullable=*/false,
+ /*is_immutable=*/false,
/*read_default=*/nullptr) },
1);
ASSERT_FALSE(s.ok());
diff --git a/src/kudu/common/schema.cc b/src/kudu/common/schema.cc
index 7f8e9ed20..48b14ae38 100644
--- a/src/kudu/common/schema.cc
+++ b/src/kudu/common/schema.cc
@@ -166,10 +166,11 @@ string ColumnSchema::ToString(ToStringMode mode) const {
string ColumnSchema::TypeToString() const {
string type_name = type_info_->name();
ToUpperCase(type_name, &type_name);
- return Substitute("$0$1 $2",
+ return Substitute("$0$1 $2$3",
type_name,
type_attributes().ToStringForType(type_info()->type()),
- is_nullable_ ? "NULLABLE" : "NOT NULL");
+ is_nullable_ ? "NULLABLE" : "NOT NULL",
+ is_immutable_ ? " IMMUTABLE" : "");
}
string ColumnSchema::AttrToString() const {
@@ -593,10 +594,21 @@ Status SchemaBuilder::AddColumn(const string& name,
bool is_nullable,
const void* read_default,
const void* write_default) {
+ return AddColumn(name, type, is_nullable, false, read_default, write_default);
+}
+
+Status SchemaBuilder::AddColumn(const std::string& name,
+ DataType type,
+ bool is_nullable,
+ bool is_immutable,
+ const void* read_default,
+ const void* write_default) {
if (name.empty()) {
return Status::InvalidArgument("column name must be non-empty");
}
- return AddColumn(ColumnSchema(name, type, is_nullable, read_default, write_default), false);
+
+ return AddColumn(ColumnSchema(name, type, is_nullable, is_immutable,
+ read_default, write_default), false);
}
Status SchemaBuilder::RemoveColumn(const string& name) {
diff --git a/src/kudu/common/schema.h b/src/kudu/common/schema.h
index 4667daee2..df1d725f3 100644
--- a/src/kudu/common/schema.h
+++ b/src/kudu/common/schema.h
@@ -48,6 +48,7 @@
namespace kudu {
class Schema;
+
typedef std::shared_ptr<Schema> SchemaPtr;
} // namespace kudu
@@ -66,7 +67,6 @@ typedef std::shared_ptr<Schema> SchemaPtr;
<< (s2).ToString(); \
} while (0)
-template <class X> struct GoodFastHash;
namespace kudu {
@@ -208,6 +208,8 @@ class ColumnSchema {
// name: column name
// type: column type (e.g. UINT8, INT32, STRING, ...)
// is_nullable: true if a row value can be null
+ // is_immutable: true if the column is immutable.
+ // Immutable column means the cell value can not be updated after the first insert.
// read_default: default value used on read if the column was not present before alter.
// The value will be copied and released on ColumnSchema destruction.
// write_default: default value added to the row if the column value was
@@ -218,13 +220,15 @@ class ColumnSchema {
// Example:
// ColumnSchema col_a("a", UINT32)
// ColumnSchema col_b("b", STRING, true);
+ // ColumnSchema col_b("b", STRING, false, true);
// uint32_t default_i32 = -15;
- // ColumnSchema col_c("c", INT32, false, &default_i32);
+ // ColumnSchema col_c("c", INT32, false, false, &default_i32);
// Slice default_str("Hello");
- // ColumnSchema col_d("d", STRING, false, &default_str);
+ // ColumnSchema col_d("d", STRING, false, false, &default_str);
ColumnSchema(std::string name,
DataType type,
bool is_nullable = false,
+ bool is_immutable = false,
const void* read_default = nullptr,
const void* write_default = nullptr,
ColumnStorageAttributes attributes = ColumnStorageAttributes(),
@@ -233,6 +237,7 @@ class ColumnSchema {
: name_(std::move(name)),
type_info_(GetTypeInfo(type)),
is_nullable_(is_nullable),
+ is_immutable_(is_immutable),
read_default_(read_default ? std::make_shared<Variant>(type, read_default) : nullptr),
attributes_(attributes),
type_attributes_(type_attributes),
@@ -252,6 +257,10 @@ class ColumnSchema {
return is_nullable_;
}
+ bool is_immutable() const {
+ return is_immutable_;
+ }
+
const std::string& name() const {
return name_;
}
@@ -327,6 +336,7 @@ class ColumnSchema {
bool EqualsType(const ColumnSchema& other) const {
if (this == &other) return true;
return is_nullable_ == other.is_nullable_ &&
+ is_immutable_ == other.is_immutable_ &&
type_info()->type() == other.type_info()->type() &&
type_attributes().EqualsForType(other.type_attributes(), type_info()->type());
}
@@ -436,6 +446,7 @@ class ColumnSchema {
std::string name_;
const TypeInfo* type_info_;
bool is_nullable_;
+ bool is_immutable_;
// use shared_ptr since the ColumnSchema is always copied around.
std::shared_ptr<Variant> read_default_;
std::shared_ptr<Variant> write_default_;
@@ -1044,11 +1055,11 @@ class SchemaBuilder {
Status AddColumn(const ColumnSchema& column, bool is_key);
Status AddColumn(const std::string& name, DataType type) {
- return AddColumn(name, type, false, nullptr, nullptr);
+ return AddColumn(name, type, false, false, nullptr, nullptr);
}
Status AddNullableColumn(const std::string& name, DataType type) {
- return AddColumn(name, type, true, nullptr, nullptr);
+ return AddColumn(name, type, true, false, nullptr, nullptr);
}
Status AddColumn(const std::string& name,
@@ -1057,6 +1068,13 @@ class SchemaBuilder {
const void* read_default,
const void* write_default);
+ Status AddColumn(const std::string& name,
+ DataType type,
+ bool is_nullable,
+ bool is_immutable,
+ const void* read_default,
+ const void* write_default);
+
Status RemoveColumn(const std::string& name);
Status RenameColumn(const std::string& old_name, const std::string& new_name);
diff --git a/src/kudu/common/wire_protocol-test.cc b/src/kudu/common/wire_protocol-test.cc
index 5e7333c16..175866c7b 100644
--- a/src/kudu/common/wire_protocol-test.cc
+++ b/src/kudu/common/wire_protocol-test.cc
@@ -722,7 +722,7 @@ TEST_F(WireProtocolTest, TestColumnDefaultValue) {
ASSERT_FALSE(col1fpb->has_write_default());
ASSERT_TRUE(col1fpb->read_default_value() == nullptr);
- ColumnSchema col2("col2", STRING, false, &read_default_str);
+ ColumnSchema col2("col2", STRING, false, false, &read_default_str);
ColumnSchemaToPB(col2, &pb);
optional<ColumnSchema> col2fpb;
ASSERT_OK(ColumnSchemaFromPB(pb, &col2fpb));
@@ -731,7 +731,7 @@ TEST_F(WireProtocolTest, TestColumnDefaultValue) {
ASSERT_EQ(read_default_str, *static_cast<const Slice *>(col2fpb->read_default_value()));
ASSERT_EQ(nullptr, static_cast<const Slice *>(col2fpb->write_default_value()));
- ColumnSchema col3("col3", STRING, false, &read_default_str, &write_default_str);
+ ColumnSchema col3("col3", STRING, false, false, &read_default_str, &write_default_str);
ColumnSchemaToPB(col3, &pb);
optional<ColumnSchema> col3fpb;
ASSERT_OK(ColumnSchemaFromPB(pb, &col3fpb));
@@ -740,7 +740,7 @@ TEST_F(WireProtocolTest, TestColumnDefaultValue) {
ASSERT_EQ(read_default_str, *static_cast<const Slice *>(col3fpb->read_default_value()));
ASSERT_EQ(write_default_str, *static_cast<const Slice *>(col3fpb->write_default_value()));
- ColumnSchema col4("col4", UINT32, false, &read_default_u32);
+ ColumnSchema col4("col4", UINT32, false, false, &read_default_u32);
ColumnSchemaToPB(col4, &pb);
optional<ColumnSchema> col4fpb;
ASSERT_OK(ColumnSchemaFromPB(pb, &col4fpb));
@@ -749,7 +749,7 @@ TEST_F(WireProtocolTest, TestColumnDefaultValue) {
ASSERT_EQ(read_default_u32, *static_cast<const uint32_t *>(col4fpb->read_default_value()));
ASSERT_EQ(nullptr, static_cast<const uint32_t *>(col4fpb->write_default_value()));
- ColumnSchema col5("col5", UINT32, false, &read_default_u32, &write_default_u32);
+ ColumnSchema col5("col5", UINT32, false, false, &read_default_u32, &write_default_u32);
ColumnSchemaToPB(col5, &pb);
optional<ColumnSchema> col5fpb;
ASSERT_OK(ColumnSchemaFromPB(pb, &col5fpb));
diff --git a/src/kudu/common/wire_protocol.cc b/src/kudu/common/wire_protocol.cc
index 434854e40..aa10d9a0c 100644
--- a/src/kudu/common/wire_protocol.cc
+++ b/src/kudu/common/wire_protocol.cc
@@ -32,7 +32,6 @@
#include <glog/logging.h>
#include <google/protobuf/map.h>
#include <google/protobuf/stubs/common.h>
-#include <google/protobuf/stubs/port.h>
#include "kudu/common/column_predicate.h"
#include "kudu/common/columnblock.h"
@@ -52,7 +51,6 @@
#include "kudu/gutil/walltime.h"
#include "kudu/util/bitmap.h"
#include "kudu/util/block_bloom_filter.h"
-#include "kudu/util/compression/compression.pb.h"
#include "kudu/util/faststring.h"
#include "kudu/util/memory/arena.h"
#include "kudu/util/net/net_util.h"
@@ -120,6 +118,8 @@ void StatusToPB(const Status& status, AppStatusPB* pb) {
pb->set_code(AppStatusPB::INCOMPLETE);
} else if (status.IsEndOfFile()) {
pb->set_code(AppStatusPB::END_OF_FILE);
+ } else if (status.IsImmutable()) {
+ pb->set_code(AppStatusPB::IMMUTABLE);
} else {
LOG(WARNING) << "Unknown error code translation from internal error "
<< status.ToString() << ": sending UNKNOWN_ERROR";
@@ -183,6 +183,8 @@ Status StatusFromPB(const AppStatusPB& pb) {
return Status::Incomplete(pb.message(), "", posix_code);
case AppStatusPB::END_OF_FILE:
return Status::EndOfFile(pb.message(), "", posix_code);
+ case AppStatusPB::IMMUTABLE:
+ return Status::Immutable(pb.message(), "", posix_code);
case AppStatusPB::UNKNOWN_ERROR:
default:
LOG(WARNING) << "Unknown error code in status: " << SecureShortDebugString(pb);
@@ -230,6 +232,7 @@ void ColumnSchemaToPB(const ColumnSchema& col_schema, ColumnSchemaPB *pb, int fl
pb->Clear();
pb->set_name(col_schema.name());
pb->set_is_nullable(col_schema.is_nullable());
+ pb->set_immutable(col_schema.is_immutable());
DataType type = col_schema.type_info()->type();
pb->set_type(type);
// Only serialize precision and scale for decimal types.
@@ -331,7 +334,9 @@ Status ColumnSchemaFromPB(const ColumnSchemaPB& pb, optional<ColumnSchema>* col_
// in protobuf is the empty string. So, it's safe to use pb.comment() directly
// regardless of whether has_comment() is true or false.
// https://developers.google.com/protocol-buffers/docs/proto#optional
+ bool immutable = pb.has_immutable() ? pb.immutable() : false;
*col_schema = ColumnSchema(pb.name(), pb.type(), pb.is_nullable(),
+ immutable,
read_default_ptr, write_default_ptr,
attributes, type_attributes, pb.comment());
return Status::OK();
diff --git a/src/kudu/common/wire_protocol.proto b/src/kudu/common/wire_protocol.proto
index b6b0c6921..c019eab9c 100644
--- a/src/kudu/common/wire_protocol.proto
+++ b/src/kudu/common/wire_protocol.proto
@@ -57,6 +57,7 @@ message AppStatusPB {
INCOMPLETE = 17;
END_OF_FILE = 18;
CANCELLED = 19;
+ IMMUTABLE = 20;
}
required ErrorCode code = 1;
diff --git a/src/kudu/consensus/log-test.cc b/src/kudu/consensus/log-test.cc
index 46977fb56..c1005b104 100644
--- a/src/kudu/consensus/log-test.cc
+++ b/src/kudu/consensus/log-test.cc
@@ -18,14 +18,15 @@
#include "kudu/consensus/log.h"
#include <algorithm>
-#include <cstddef>
#include <cerrno>
+#include <cstddef>
#include <cstdint>
#include <functional>
#include <limits>
#include <memory>
#include <ostream>
#include <string>
+#include <type_traits>
#include <vector>
#include <gflags/gflags.h>
@@ -1073,7 +1074,7 @@ TEST_F(LogTest, TestGetGCableDataSize) {
const int kNumTotalSegments = 5;
const int kNumOpsPerSegment = 5;
- const int kSegmentSizeBytes = 331 + env_->GetEncryptionHeaderSize();
+ const int64_t kSegmentSizeBytes = 337 + env_->GetEncryptionHeaderSize();
OpId op_id = MakeOpId(1, 10);
// Create 5 segments, starting from log index 10, with 5 ops per segment.
// [10-14], [15-19], [20-24], [25-29], [30-34]
diff --git a/src/kudu/master/master.proto b/src/kudu/master/master.proto
index d556245fa..13050c30f 100644
--- a/src/kudu/master/master.proto
+++ b/src/kudu/master/master.proto
@@ -1138,6 +1138,8 @@ enum MasterFeatures {
IGNORE_OPERATIONS = 7;
// Whether master supports tables with range-specific hash schemas.
RANGE_SPECIFIC_HASH_SCHEMA = 8;
+ // Similar to IGNORE_OPERATIONS, but this is for UPSERT_IGNORE specifically.
+ UPSERT_IGNORE = 9;
}
service MasterService {
diff --git a/src/kudu/master/master_service.cc b/src/kudu/master/master_service.cc
index d7c3f6ca8..e2eaa3cbb 100644
--- a/src/kudu/master/master_service.cc
+++ b/src/kudu/master/master_service.cc
@@ -21,6 +21,7 @@
#include <optional>
#include <ostream>
#include <string>
+#include <type_traits>
#include <unordered_map>
#include <utility>
#include <vector>
@@ -102,10 +103,16 @@ TAG_FLAG(master_support_change_config, advanced);
TAG_FLAG(master_support_change_config, runtime);
DEFINE_bool(master_support_ignore_operations, true,
- "Whether the cluster supports support ignore operations.");
+ "Whether the cluster supports ignore operations, including "
+ "INSERT_IGNORE, DELETE_IGNORE and UPDATE_IGNORE).");
TAG_FLAG(master_support_ignore_operations, hidden);
TAG_FLAG(master_support_ignore_operations, runtime);
+DEFINE_bool(master_support_upsert_ignore_operations, true,
+ "Whether the cluster supports UPSERT_IGNORE operations.");
+TAG_FLAG(master_support_upsert_ignore_operations, hidden);
+TAG_FLAG(master_support_upsert_ignore_operations, runtime);
+
using google::protobuf::Message;
using kudu::consensus::ReplicaManagementInfoPB;
@@ -886,6 +893,8 @@ bool MasterServiceImpl::SupportsFeature(uint32_t feature) const {
return FLAGS_master_support_ignore_operations;
case MasterFeatures::RANGE_SPECIFIC_HASH_SCHEMA:
return FLAGS_enable_per_range_hash_schemas;
+ case MasterFeatures::UPSERT_IGNORE:
+ return FLAGS_master_support_upsert_ignore_operations;
default:
return false;
}
diff --git a/src/kudu/tablet/all_types-scan-correctness-test.cc b/src/kudu/tablet/all_types-scan-correctness-test.cc
index 5340e0443..859a3194c 100644
--- a/src/kudu/tablet/all_types-scan-correctness-test.cc
+++ b/src/kudu/tablet/all_types-scan-correctness-test.cc
@@ -79,9 +79,9 @@ static const int kStrlen = 10;
struct RowOpsBase {
RowOpsBase(DataType type, EncodingType encoding) : type_(type), encoding_(encoding) {
schema_ = Schema({ColumnSchema("key", INT32),
- ColumnSchema("val_a", type, true, nullptr, nullptr,
+ ColumnSchema("val_a", type, true, false, nullptr, nullptr,
ColumnStorageAttributes(encoding, DEFAULT_COMPRESSION)),
- ColumnSchema("val_b", type, true, nullptr, nullptr,
+ ColumnSchema("val_b", type, true, false, nullptr, nullptr,
ColumnStorageAttributes(encoding, DEFAULT_COMPRESSION))}, 1);
}
@@ -368,11 +368,11 @@ public:
ASSERT_OK(builder.AddColumn("val_c", rowops_.type_, true, default_ptr, nullptr));
AlterSchema(builder.Build());
altered_schema_ = Schema({ColumnSchema("key", INT32),
- ColumnSchema("val_a", rowops_.type_, true, nullptr, nullptr,
+ ColumnSchema("val_a", rowops_.type_, true, false, nullptr, nullptr,
ColumnStorageAttributes(rowops_.encoding_, DEFAULT_COMPRESSION)),
- ColumnSchema("val_b", rowops_.type_, true, nullptr, nullptr,
+ ColumnSchema("val_b", rowops_.type_, true, false, nullptr, nullptr,
ColumnStorageAttributes(rowops_.encoding_, DEFAULT_COMPRESSION)),
- ColumnSchema("val_c", rowops_.type_, true, default_ptr, nullptr,
+ ColumnSchema("val_c", rowops_.type_, true, false, default_ptr, nullptr,
ColumnStorageAttributes(rowops_.encoding_, DEFAULT_COMPRESSION))}, 1);
}
diff --git a/src/kudu/tablet/cfile_set-test.cc b/src/kudu/tablet/cfile_set-test.cc
index 0e6723d87..d50924488 100644
--- a/src/kudu/tablet/cfile_set-test.cc
+++ b/src/kudu/tablet/cfile_set-test.cc
@@ -24,6 +24,7 @@
#include <memory>
#include <ostream>
#include <string>
+#include <type_traits>
#include <vector>
#include <gflags/gflags.h>
@@ -88,7 +89,8 @@ class TestCFileSet : public KuduRowSetTest {
public:
TestCFileSet() :
KuduRowSetTest(Schema({ ColumnSchema("c0", INT32),
- ColumnSchema("c1", INT32, false, nullptr, nullptr, GetRLEStorage()),
+ ColumnSchema("c1", INT32, false, false,
+ nullptr, nullptr, GetRLEStorage()),
ColumnSchema("c2", INT32, true) }, 1))
{}
diff --git a/src/kudu/tablet/diskrowset-test.cc b/src/kudu/tablet/diskrowset-test.cc
index f62e4966d..a813f0395 100644
--- a/src/kudu/tablet/diskrowset-test.cc
+++ b/src/kudu/tablet/diskrowset-test.cc
@@ -25,6 +25,7 @@
#include <ostream>
#include <string>
#include <tuple>
+#include <type_traits>
#include <unordered_set>
#include <vector>
@@ -54,13 +55,10 @@
#include "kudu/tablet/delta_key.h"
#include "kudu/tablet/delta_store.h"
#include "kudu/tablet/delta_tracker.h"
-#include "kudu/tablet/deltamemstore.h"
#include "kudu/tablet/diskrowset-test-base.h"
#include "kudu/tablet/mvcc.h"
#include "kudu/tablet/rowset.h"
-#include "kudu/tablet/rowset_metadata.h"
#include "kudu/tablet/tablet-test-util.h"
-#include "kudu/tablet/tablet.h"
#include "kudu/tablet/tablet.pb.h"
#include "kudu/tablet/tablet_mem_trackers.h"
#include "kudu/util/bloom_filter.h"
@@ -73,6 +71,12 @@
#include "kudu/util/test_macros.h"
#include "kudu/util/test_util.h"
+namespace kudu {
+namespace tablet {
+class RowSetMetadata;
+} // namespace tablet
+} // namespace kudu
+
DEFINE_double(update_fraction, 0.1f, "fraction of rows to update");
DECLARE_bool(cfile_lazy_open);
DECLARE_bool(crash_on_eio);
@@ -787,7 +791,7 @@ TEST_P(DiffScanRowSetTest, TestFuzz) {
if (add_vc_is_deleted) {
bool read_default = false;
col_schemas.emplace_back("is_deleted", IS_DELETED, /*is_nullable=*/ false,
- &read_default);
+ /*is_immutable=*/ false, &read_default);
col_ids.emplace_back(schema_.max_col_id() + 1);
}
Schema projection(col_schemas, col_ids, 1);
diff --git a/src/kudu/tablet/local_tablet_writer.h b/src/kudu/tablet/local_tablet_writer.h
index f7dfcfe59..b84744eb7 100644
--- a/src/kudu/tablet/local_tablet_writer.h
+++ b/src/kudu/tablet/local_tablet_writer.h
@@ -75,6 +75,10 @@ class LocalTabletWriter {
return Write(RowOperationsPB::UPSERT, row);
}
+ Status UpsertIgnore(const KuduPartialRow& row) {
+ return Write(RowOperationsPB::UPSERT_IGNORE, row);
+ }
+
Status Delete(const KuduPartialRow& row) {
return Write(RowOperationsPB::DELETE, row);
}
diff --git a/src/kudu/tablet/ops/op.cc b/src/kudu/tablet/ops/op.cc
index 27c4464e5..c871281f8 100644
--- a/src/kudu/tablet/ops/op.cc
+++ b/src/kudu/tablet/ops/op.cc
@@ -78,6 +78,7 @@ OpMetrics::OpMetrics()
: successful_inserts(0),
insert_ignore_errors(0),
successful_upserts(0),
+ upsert_ignore_errors(0),
successful_updates(0),
update_ignore_errors(0),
successful_deletes(0),
@@ -89,6 +90,7 @@ void OpMetrics::Reset() {
successful_inserts = 0;
insert_ignore_errors = 0;
successful_upserts = 0;
+ upsert_ignore_errors = 0;
successful_updates = 0;
update_ignore_errors = 0;
successful_deletes = 0;
diff --git a/src/kudu/tablet/ops/op.h b/src/kudu/tablet/ops/op.h
index 155b003f1..10093ab92 100644
--- a/src/kudu/tablet/ops/op.h
+++ b/src/kudu/tablet/ops/op.h
@@ -62,6 +62,7 @@ struct OpMetrics {
int successful_inserts;
int insert_ignore_errors;
int successful_upserts;
+ int upsert_ignore_errors;
int successful_updates;
int update_ignore_errors;
int successful_deletes;
diff --git a/src/kudu/tablet/ops/write_op.cc b/src/kudu/tablet/ops/write_op.cc
index 6a8063f7f..7ee5af815 100644
--- a/src/kudu/tablet/ops/write_op.cc
+++ b/src/kudu/tablet/ops/write_op.cc
@@ -114,6 +114,7 @@ void AddWritePrivilegesForRowOperations(const RowOperationsPB::Type& op_type,
InsertIfNotPresent(privileges, WritePrivilegeType::INSERT);
break;
case RowOperationsPB::UPSERT:
+ case RowOperationsPB::UPSERT_IGNORE:
InsertIfNotPresent(privileges, WritePrivilegeType::INSERT);
InsertIfNotPresent(privileges, WritePrivilegeType::UPDATE);
break;
@@ -317,6 +318,7 @@ void WriteOp::Finish(OpResult result) {
metrics->rows_inserted->IncrementBy(op_m.successful_inserts);
metrics->insert_ignore_errors->IncrementBy(op_m.insert_ignore_errors);
metrics->rows_upserted->IncrementBy(op_m.successful_upserts);
+ metrics->upsert_ignore_errors->IncrementBy(op_m.upsert_ignore_errors);
metrics->rows_updated->IncrementBy(op_m.successful_updates);
metrics->update_ignore_errors->IncrementBy(op_m.update_ignore_errors);
metrics->rows_deleted->IncrementBy(op_m.successful_deletes);
@@ -504,6 +506,13 @@ void WriteOpState::UpdateMetricsForOp(const RowOp& op) {
DCHECK(!op.error_ignored);
op_metrics_.successful_upserts++;
break;
+ case RowOperationsPB::UPSERT_IGNORE:
+ if (op.error_ignored) {
+ op_metrics_.upsert_ignore_errors++;
+ } else {
+ op_metrics_.successful_upserts++;
+ }
+ break;
case RowOperationsPB::UPDATE:
DCHECK(!op.error_ignored);
op_metrics_.successful_updates++;
@@ -669,6 +678,7 @@ void WriteOpState::FillResponseMetrics(consensus::DriverType type) {
resp_metrics->set_successful_inserts(op_m.successful_inserts);
resp_metrics->set_insert_ignore_errors(op_m.insert_ignore_errors);
resp_metrics->set_successful_upserts(op_m.successful_upserts);
+ resp_metrics->set_upsert_ignore_errors(op_m.upsert_ignore_errors);
resp_metrics->set_successful_updates(op_m.successful_updates);
resp_metrics->set_update_ignore_errors(op_m.update_ignore_errors);
resp_metrics->set_successful_deletes(op_m.successful_deletes);
diff --git a/src/kudu/tablet/ops/write_op.h b/src/kudu/tablet/ops/write_op.h
index 258df2128..9bf4e6603 100644
--- a/src/kudu/tablet/ops/write_op.h
+++ b/src/kudu/tablet/ops/write_op.h
@@ -23,10 +23,10 @@
#include <mutex>
#include <optional>
#include <string>
+#include <type_traits>
#include <vector>
#include <glog/logging.h>
-#include <google/protobuf/stubs/port.h>
#include "kudu/common/row_operations.h"
#include "kudu/common/row_operations.pb.h"
diff --git a/src/kudu/tablet/row_op.cc b/src/kudu/tablet/row_op.cc
index 6a2e6617b..f112f5ba5 100644
--- a/src/kudu/tablet/row_op.cc
+++ b/src/kudu/tablet/row_op.cc
@@ -41,6 +41,7 @@ RowOp::RowOp(google::protobuf::Arena* pb_arena, DecodedRowOperation op)
if (!decoded_op.result.ok()) {
SetFailed(decoded_op.result);
}
+ error_ignored = op.error_ignored;
}
void RowOp::SetFailed(const Status& s) {
diff --git a/src/kudu/tablet/tablet-decoder-eval-test.cc b/src/kudu/tablet/tablet-decoder-eval-test.cc
index 86f5b02b3..b64f76f56 100644
--- a/src/kudu/tablet/tablet-decoder-eval-test.cc
+++ b/src/kudu/tablet/tablet-decoder-eval-test.cc
@@ -22,6 +22,7 @@
#include <memory>
#include <ostream>
#include <string>
+#include <type_traits>
#include <unordered_map>
#include <gflags/gflags.h>
@@ -41,7 +42,6 @@
#include "kudu/gutil/strings/substitute.h"
#include "kudu/tablet/local_tablet_writer.h"
#include "kudu/tablet/tablet-test-util.h"
-#include "kudu/tablet/tablet.h"
#include "kudu/util/compression/compression.pb.h"
#include "kudu/util/memory/arena.h"
#include "kudu/util/slice.h"
@@ -75,11 +75,11 @@ class TabletDecoderEvalTest : public KuduTabletTest,
public:
TabletDecoderEvalTest()
: KuduTabletTest(Schema({ColumnSchema("key", INT32),
- ColumnSchema("string_val_a", STRING, true,
+ ColumnSchema("string_val_a", STRING, true, false,
nullptr, nullptr,
ColumnStorageAttributes(DICT_ENCODING,
DEFAULT_COMPRESSION)),
- ColumnSchema("string_val_b", STRING, true,
+ ColumnSchema("string_val_b", STRING, true, false,
nullptr, nullptr,
ColumnStorageAttributes(DICT_ENCODING,
DEFAULT_COMPRESSION))}, 1))
diff --git a/src/kudu/tablet/tablet-test-base.h b/src/kudu/tablet/tablet-test-base.h
index 8cad7cb2a..a275fdecf 100644
--- a/src/kudu/tablet/tablet-test-base.h
+++ b/src/kudu/tablet/tablet-test-base.h
@@ -291,6 +291,54 @@ struct NullableValueTestSetup {
}
};
+// Setup for testing nullable columns
+struct ImmutableColumnTestSetup {
+ static Schema CreateSchema() {
+ return Schema({ ColumnSchema("key", INT32),
+ ColumnSchema("key_idx", INT32),
+ ColumnSchema("val", INT32),
+ ColumnSchema("imm_val", INT32, true, true) }, 1);
+ }
+
+ static void BuildRowKey(KuduPartialRow *row, int64_t i) {
+ CHECK_OK(row->SetInt32(0, (int32_t)i));
+ }
+
+ // builds a row key from an existing row for updates
+ template<class RowType>
+ void BuildRowKeyFromExistingRow(KuduPartialRow *row, const RowType& src_row) {
+ CHECK_OK(row->SetInt32(0, *reinterpret_cast<const int32_t*>(src_row.cell_ptr(0))));
+ }
+
+ static void BuildRow(KuduPartialRow *row, int64_t key_idx,
+ int32_t val = 0, bool set_immutable_column = false) {
+ BuildRowKey(row, key_idx);
+ CHECK_OK(row->SetInt32(1, key_idx));
+ CHECK_OK(row->SetInt32(2, val));
+ if (set_immutable_column) {
+ CHECK_OK(row->SetInt32(3, val));
+ } else {
+ CHECK_OK(row->Unset(3));
+ }
+ }
+
+ static std::string FormatDebugRow(int64_t key_idx, int64_t val, bool /* updated */,
+ std::optional<int64_t> immutable_column_val = std::nullopt) {
+ if (immutable_column_val) {
+ return strings::Substitute(
+ "(int32 key=$0, int32 key_idx=$1, int32 val=$2, int32 imm_val=$3)",
+ static_cast<int32_t>(key_idx), key_idx, val, *immutable_column_val);
+ }
+ return strings::Substitute(
+ "(int32 key=$0, int32 key_idx=$1, int32 val=$2, int32 imm_val=NULL)",
+ static_cast<int32_t>(key_idx), key_idx, val);
+ }
+
+ static uint64_t GetMaxRows() {
+ return std::numeric_limits<uint32_t>::max() - 1;
+ }
+};
+
// Use this with TYPED_TEST_SUITE from gtest
typedef ::testing::Types<
StringKeyTestSetup,
@@ -298,9 +346,12 @@ typedef ::testing::Types<
IntKeyTestSetup<INT16>,
IntKeyTestSetup<INT32>,
IntKeyTestSetup<INT64>,
- NullableValueTestSetup
+ NullableValueTestSetup,
+ ImmutableColumnTestSetup
> TabletTestHelperTypes;
+typedef ::testing::Types<ImmutableColumnTestSetup> TestImmutableColumnHelperTypes;
+
template<class TESTSETUP>
class TabletTestBase : public KuduTabletTest {
public:
@@ -337,6 +388,14 @@ class TabletTestBase : public KuduTabletTest {
InsertOrUpsertTestRows(RowOperationsPB::UPSERT, first_row, count, val, ts);
}
+ // Upserts "count" rows, ignoring immutable column errors.
+ void UpsertIgnoreTestRows(int64_t first_row,
+ int64_t count,
+ int32_t val,
+ TimeSeries *ts = nullptr) {
+ InsertOrUpsertTestRows(RowOperationsPB::UPSERT_IGNORE, first_row, count, val, ts);
+ }
+
// Deletes 'count' rows, starting with 'first_row'.
void DeleteTestRows(int64_t first_row, int64_t count) {
LocalTabletWriter writer(tablet().get(), &client_schema_);
@@ -362,6 +421,8 @@ class TabletTestBase : public KuduTabletTest {
CHECK_OK(writer.InsertIgnore(row));
} else if (type == RowOperationsPB::UPSERT) {
CHECK_OK(writer.Upsert(row));
+ } else if (type == RowOperationsPB::UPSERT_IGNORE) {
+ CHECK_OK(writer.UpsertIgnore(row));
} else {
LOG(FATAL) << "bad type: " << type;
}
diff --git a/src/kudu/tablet/tablet-test-util.h b/src/kudu/tablet/tablet-test-util.h
index ad3f277f6..62cd8b5f2 100644
--- a/src/kudu/tablet/tablet-test-util.h
+++ b/src/kudu/tablet/tablet-test-util.h
@@ -792,7 +792,7 @@ static inline Schema GetRandomProjection(const Schema& schema,
if (allow == AllowIsDeleted::YES && prng->Uniform(10) == 0) {
bool read_default = false;
projected_cols.emplace_back("is_deleted", IS_DELETED, /*is_nullable=*/ false,
- &read_default);
+ /*is_immutable=*/ false, &read_default);
projected_col_ids.emplace_back(schema.max_col_id() + 1);
}
return Schema(projected_cols, projected_col_ids, 0);
diff --git a/src/kudu/tablet/tablet-test.cc b/src/kudu/tablet/tablet-test.cc
index 2e592de53..20217318d 100644
--- a/src/kudu/tablet/tablet-test.cc
+++ b/src/kudu/tablet/tablet-test.cc
@@ -25,6 +25,7 @@
#include <optional>
#include <ostream>
#include <string>
+#include <type_traits>
#include <utility>
#include <vector>
@@ -57,7 +58,7 @@
#include "kudu/tablet/mvcc.h"
#include "kudu/tablet/rowset.h"
#include "kudu/tablet/rowset_metadata.h"
-#include "kudu/tablet/rowset_tree.h"
+#include "kudu/tablet/rowset_tree.h" // IWYU pragma: keep
#include "kudu/tablet/tablet-test-base.h"
#include "kudu/tablet/tablet-test-util.h"
#include "kudu/tablet/tablet.pb.h"
@@ -337,12 +338,12 @@ TYPED_TEST(TestTablet, TestReinserts) {
<< snaps[0].ToString();
ASSERT_EQ(expected_rows[1]->size(), 1) << "Got the wrong result from snap: "
<< snaps[1].ToString();
- ASSERT_STR_CONTAINS((*expected_rows[1])[0], "int32 key_idx=1, int32 val=0)");
+ ASSERT_STR_CONTAINS((*expected_rows[1])[0], "int32 key_idx=1, int32 val=0");
ASSERT_EQ(expected_rows[2]->size(), 0) << "Got the wrong result from snap: "
<< snaps[2].ToString();
ASSERT_EQ(expected_rows[3]->size(), 1) << "Got the wrong result from snap: "
<< snaps[3].ToString();
- ASSERT_STR_CONTAINS((*expected_rows[3])[0], "int32 key_idx=1, int32 val=1)");
+ ASSERT_STR_CONTAINS((*expected_rows[3])[0], "int32 key_idx=1, int32 val=1");
ASSERT_EQ(expected_rows[4]->size(), 0) << "Got the wrong result from snap: "
<< snaps[4].ToString();
NO_FATALS(this->CheckLiveRowsCount(0));
@@ -892,6 +893,90 @@ TYPED_TEST(TestTablet, TestUpsert) {
NO_FATALS(this->CheckLiveRowsCount(1));
}
+TYPED_TEST(TestTablet, TestUpsertIgnore) {
+ vector<string> rows;
+ const auto& upserts_as_updates = this->tablet()->metrics()->upserts_as_updates;
+
+ // Upsert a new row.
+ this->UpsertIgnoreTestRows(0, 1, 1000);
+ EXPECT_EQ(0, upserts_as_updates->value());
+
+ // Upsert a row that is in the MRS.
+ this->UpsertIgnoreTestRows(0, 1, 1001);
+ EXPECT_EQ(1, upserts_as_updates->value());
+
+ ASSERT_OK(this->IterateToStringList(&rows));
+ EXPECT_EQ(vector<string>{ this->setup_.FormatDebugRow(0, 1001, false) }, rows);
+
+ // Flush it.
+ ASSERT_OK(this->tablet()->Flush());
+ ASSERT_OK(this->IterateToStringList(&rows));
+ EXPECT_EQ(vector<string>{ this->setup_.FormatDebugRow(0, 1001, false) }, rows);
+
+ // Upsert a row that is in the DRS.
+ this->UpsertIgnoreTestRows(0, 1, 1002);
+ EXPECT_EQ(2, upserts_as_updates->value());
+ ASSERT_OK(this->IterateToStringList(&rows));
+ EXPECT_EQ(vector<string>{ this->setup_.FormatDebugRow(0, 1002, false) }, rows);
+ NO_FATALS(this->CheckLiveRowsCount(1));
+}
+
+template<class SETUP>
+class TestImmutableColumn : public TabletTestBase<SETUP> {
+ typedef SETUP Type;
+
+ public:
+ // Verify that iteration doesn't fail
+ void CheckCanIterate() {
+ vector<string> out_rows;
+ ASSERT_OK(this->IterateToStringList(&out_rows));
+ }
+
+ void CheckLiveRowsCount(int64_t expect) {
+ uint64_t count = 0;
+ ASSERT_OK(this->tablet()->CountLiveRows(&count));
+ ASSERT_EQ(expect, count);
+ }
+};
+TYPED_TEST_SUITE(TestImmutableColumn, TestImmutableColumnHelperTypes);
+
+TYPED_TEST(TestImmutableColumn, TestUpsert) {
+ vector<string> rows;
+ LocalTabletWriter writer(this->tablet().get(), &this->client_schema_);
+ const auto& upserts_as_updates = this->tablet()->metrics()->upserts_as_updates;
+
+ // Upsert a new row with immutable column.
+ KuduPartialRow row(&this->client_schema_);
+ this->setup_.BuildRow(&row, 0, 1000, true /* set_immutable_column */);
+ CHECK_OK(writer.Upsert(row));
+ EXPECT_EQ(0, upserts_as_updates->value());
+ ASSERT_OK(this->IterateToStringList(&rows));
+ EXPECT_EQ(vector<string>{this->setup_.FormatDebugRow(0, 1000, false, 1000) }, rows);
+
+ // Upsert the same row without immutable columns.
+ this->setup_.BuildRow(&row, 0, 1001, false /* set_immutable_column */);
+ CHECK_OK(writer.Upsert(row));
+ EXPECT_EQ(1, upserts_as_updates->value());
+ ASSERT_OK(this->IterateToStringList(&rows));
+ EXPECT_EQ(vector<string>{this->setup_.FormatDebugRow(0, 1001, false, 1000) }, rows);
+
+ // Upsert the same row with immutable columns.
+ this->setup_.BuildRow(&row, 0, 1002, true /* set_immutable_column */);
+ Status s = writer.Upsert(row);
+ EXPECT_TRUE(s.IsImmutable());
+ ASSERT_STR_CONTAINS(s.ToString(), "UPDATE not allowed for immutable column: imm_val");
+ EXPECT_EQ(1, upserts_as_updates->value());
+ ASSERT_OK(this->IterateToStringList(&rows));
+ EXPECT_EQ(vector<string>{this->setup_.FormatDebugRow(0, 1001, false, 1000) }, rows);
+
+ // UpsertIgnore the same row with immutable columns.
+ this->setup_.BuildRow(&row, 0, 1003, true /* set_immutable_column */);
+ CHECK_OK(writer.UpsertIgnore(row));
+ EXPECT_EQ(2, upserts_as_updates->value());
+ ASSERT_OK(this->IterateToStringList(&rows));
+ EXPECT_EQ(vector<string>{this->setup_.FormatDebugRow(0, 1003, false, 1000) }, rows);
+}
+
// Test that when a row has been updated many times, it always yields
// the most recent value.
TYPED_TEST(TestTablet, TestMultipleUpdates) {
@@ -1622,7 +1707,7 @@ TYPED_TEST(TestTablet, TestDiffScanUnobservableOperations) {
vector<ColumnSchema> col_schemas(this->client_schema().columns());
bool read_default = false;
col_schemas.emplace_back("is_deleted", IS_DELETED, /*is_nullable=*/ false,
- &read_default);
+ /*is_immutable=*/ false, &read_default);
Schema projection(col_schemas, this->client_schema().num_key_columns());
// Do the diff scan.
diff --git a/src/kudu/tablet/tablet.cc b/src/kudu/tablet/tablet.cc
index bf691c2d4..f4e5340d4 100644
--- a/src/kudu/tablet/tablet.cc
+++ b/src/kudu/tablet/tablet.cc
@@ -18,6 +18,7 @@
#include "kudu/tablet/tablet.h"
#include <algorithm>
+#include <ctime>
#include <functional>
#include <iterator>
#include <memory>
@@ -71,7 +72,6 @@
#include "kudu/tablet/ops/write_op.h"
#include "kudu/tablet/row_op.h"
#include "kudu/tablet/rowset_info.h"
-#include "kudu/tablet/rowset_metadata.h"
#include "kudu/tablet/rowset_tree.h"
#include "kudu/tablet/svg_dump.h"
#include "kudu/tablet/tablet.pb.h"
@@ -98,6 +98,12 @@
#include "kudu/util/trace.h"
#include "kudu/util/url-coding.h"
+namespace kudu {
+namespace tablet {
+class RowSetMetadata;
+} // namespace tablet
+} // namespace kudu
+
DEFINE_bool(prevent_kudu_2233_corruption, true,
"Whether or not to prevent KUDU-2233 corruptions. Used for testing only!");
TAG_FLAG(prevent_kudu_2233_corruption, unsafe);
@@ -673,6 +679,7 @@ Status Tablet::ValidateOp(const RowOp& op) {
case RowOperationsPB::INSERT:
case RowOperationsPB::INSERT_IGNORE:
case RowOperationsPB::UPSERT:
+ case RowOperationsPB::UPSERT_IGNORE:
return ValidateInsertOrUpsertUnlocked(op);
case RowOperationsPB::UPDATE:
@@ -731,7 +738,18 @@ Status Tablet::InsertOrUpsertUnlocked(const IOContext* io_context,
if (op->present_in_rowset) {
switch (op_type) {
case RowOperationsPB::UPSERT:
- return ApplyUpsertAsUpdate(io_context, op_state, op, op->present_in_rowset, stats);
+ case RowOperationsPB::UPSERT_IGNORE: {
+ Status s = ApplyUpsertAsUpdate(io_context, op_state, op, op->present_in_rowset, stats);
+ if (s.ok()) {
+ return Status::OK();
+ }
+ if (s.IsImmutable() && op_type == RowOperationsPB::UPSERT_IGNORE) {
+ op->SetErrorIgnored();
+ return Status::OK();
+ }
+ op->SetFailed(s);
+ return s;
+ }
case RowOperationsPB::INSERT_IGNORE:
op->SetErrorIgnored();
return Status::OK();
@@ -802,7 +820,18 @@ Status Tablet::InsertOrUpsertUnlocked(const IOContext* io_context,
if (s.IsAlreadyPresent()) {
switch (op_type) {
case RowOperationsPB::UPSERT:
- return ApplyUpsertAsUpdate(io_context, op_state, op, comps->memrowset.get(), stats);
+ case RowOperationsPB::UPSERT_IGNORE: {
+ Status s = ApplyUpsertAsUpdate(io_context, op_state, op, comps->memrowset.get(), stats);
+ if (s.ok()) {
+ return Status::OK();
+ }
+ if (s.IsImmutable() && op_type == RowOperationsPB::UPSERT_IGNORE) {
+ op->SetErrorIgnored();
+ return Status::OK();
+ }
+ op->SetFailed(s);
+ return s;
+ }
case RowOperationsPB::INSERT_IGNORE:
op->SetErrorIgnored();
return Status::OK();
@@ -825,6 +854,8 @@ Status Tablet::ApplyUpsertAsUpdate(const IOContext* io_context,
RowOp* upsert,
RowSet* rowset,
ProbeStats* stats) {
+ const auto op_type = upsert->decoded_op.type;
+ bool error_ignored = false;
const auto* schema = this->schema().get();
ConstContiguousRow row(schema, upsert->decoded_op.row_data);
faststring buf;
@@ -835,6 +866,15 @@ Status Tablet::ApplyUpsertAsUpdate(const IOContext* io_context,
// values back to their defaults when unset.
if (!BitmapTest(upsert->decoded_op.isset_bitmap, i)) continue;
const auto& c = schema->column(i);
+ if (c.is_immutable()) {
+ if (op_type == RowOperationsPB::UPSERT) {
+ return Status::Immutable("UPDATE not allowed for immutable column", c.ToString());
+ }
+ DCHECK_EQ(op_type, RowOperationsPB::UPSERT_IGNORE);
+ error_ignored = true;
+ continue;
+ }
+
const void* val = c.is_nullable() ? row.nullable_cell_ptr(i) : row.cell_ptr(i);
enc.AddColumnUpdate(c, schema->column_id(i), val);
}
@@ -847,6 +887,9 @@ Status Tablet::ApplyUpsertAsUpdate(const IOContext* io_context,
op_state->pb_arena());
if (enc.is_empty()) {
upsert->SetMutateSucceeded(result);
+ if (error_ignored) {
+ upsert->error_ignored = true;
+ }
return Status::OK();
}
@@ -1296,8 +1339,9 @@ Status Tablet::ApplyRowOperation(const IOContext* io_context,
case RowOperationsPB::INSERT:
case RowOperationsPB::INSERT_IGNORE:
case RowOperationsPB::UPSERT:
+ case RowOperationsPB::UPSERT_IGNORE:
s = InsertOrUpsertUnlocked(io_context, op_state, row_op, stats);
- if (s.IsAlreadyPresent()) {
+ if (s.IsAlreadyPresent() || s.IsImmutable()) {
return Status::OK();
}
return s;
diff --git a/src/kudu/tablet/tablet.h b/src/kudu/tablet/tablet.h
index 6298c6c64..607d5aeba 100644
--- a/src/kudu/tablet/tablet.h
+++ b/src/kudu/tablet/tablet.h
@@ -617,7 +617,7 @@ class Tablet {
// Validate the given update/delete operation.
static Status ValidateMutateUnlocked(const RowOp& op);
- // Perform an INSERT, INSERT_IGNORE, or UPSERT operation, assuming that the op is
+ // Perform an INSERT, INSERT_IGNORE, UPSERT, or UPSERT_IGNORE operation, assuming that the op is
// already in a prepared state. This state ensures that:
// - the row lock is acquired
// - the tablet components have been acquired
diff --git a/src/kudu/tablet/tablet_bootstrap.cc b/src/kudu/tablet/tablet_bootstrap.cc
index 3267f0c6e..fea9b8f10 100644
--- a/src/kudu/tablet/tablet_bootstrap.cc
+++ b/src/kudu/tablet/tablet_bootstrap.cc
@@ -25,6 +25,7 @@
#include <optional>
#include <ostream>
#include <string>
+#include <type_traits>
#include <unordered_map>
#include <unordered_set>
#include <utility>
@@ -75,7 +76,7 @@
#include "kudu/tablet/ops/write_op.h"
#include "kudu/tablet/row_op.h"
#include "kudu/tablet/rowset.h"
-#include "kudu/tablet/rowset_metadata.h"
+#include "kudu/tablet/rowset_metadata.h" // IWYU pragma: keep
#include "kudu/tablet/tablet.h"
#include "kudu/tablet/tablet.pb.h"
#include "kudu/tablet/tablet_metadata.h"
@@ -1681,7 +1682,8 @@ Status TabletBootstrap::ApplyOperations(const IOContext* io_context,
switch (op->decoded_op.type) {
case RowOperationsPB::INSERT:
case RowOperationsPB::INSERT_IGNORE:
- case RowOperationsPB::UPSERT: {
+ case RowOperationsPB::UPSERT:
+ case RowOperationsPB::UPSERT_IGNORE: {
// TODO(unknown): should we have a separate counter for upserts?
stats_.inserts_seen++;
if (op->has_result()) {
diff --git a/src/kudu/tablet/tablet_metrics.cc b/src/kudu/tablet/tablet_metrics.cc
index b52065f1c..fd6a7c7bb 100644
--- a/src/kudu/tablet/tablet_metrics.cc
+++ b/src/kudu/tablet/tablet_metrics.cc
@@ -40,6 +40,15 @@ METRIC_DEFINE_counter(tablet, rows_upserted, "Rows Upserted",
kudu::MetricUnit::kRows,
"Number of rows upserted into this tablet since service start",
kudu::MetricLevel::kInfo);
+METRIC_DEFINE_counter(tablet, upsert_ignore_errors, "Upsert Ignore Errors",
+ kudu::MetricUnit::kRows,
+ "Number of upsert ignore operations for this tablet which were "
+ "ignored due to an error since service start. This metric counts "
+ "the number of attempts to update a present row by changing the "
+ "value of any of its immutable cells. Note that the rest of the "
+ "cells (i.e. the mutable ones) in such case are updated accordingly "
+ "to the operation's data.",
+ kudu::MetricLevel::kDebug);
METRIC_DEFINE_counter(tablet, rows_updated, "Rows Updated",
kudu::MetricUnit::kRows,
"Number of row update operations performed on this tablet since service start",
@@ -374,6 +383,7 @@ TabletMetrics::TabletMetrics(const scoped_refptr<MetricEntity>& entity)
MINIT(rows_updated),
MINIT(rows_deleted),
MINIT(insert_ignore_errors),
+ MINIT(upsert_ignore_errors),
MINIT(update_ignore_errors),
MINIT(delete_ignore_errors),
MINIT(insertions_failed_dup_key),
diff --git a/src/kudu/tablet/tablet_metrics.h b/src/kudu/tablet/tablet_metrics.h
index 6be2e7044..1ef983605 100644
--- a/src/kudu/tablet/tablet_metrics.h
+++ b/src/kudu/tablet/tablet_metrics.h
@@ -49,6 +49,7 @@ struct TabletMetrics {
scoped_refptr<Counter> rows_updated;
scoped_refptr<Counter> rows_deleted;
scoped_refptr<Counter> insert_ignore_errors;
+ scoped_refptr<Counter> upsert_ignore_errors;
scoped_refptr<Counter> update_ignore_errors;
scoped_refptr<Counter> delete_ignore_errors;
scoped_refptr<Counter> insertions_failed_dup_key;
diff --git a/src/kudu/tablet/tablet_random_access-test.cc b/src/kudu/tablet/tablet_random_access-test.cc
index 42315045c..9884395d8 100644
--- a/src/kudu/tablet/tablet_random_access-test.cc
+++ b/src/kudu/tablet/tablet_random_access-test.cc
@@ -121,7 +121,7 @@ class TestRandomAccess : public KuduTabletTest {
if (!cur_val) {
// If there is no row, then randomly insert, insert ignore,
// update ignore, delete ignore, or upsert.
- switch (rand() % 5) {
+ switch (rand() % 6) {
case 1:
cur_val = InsertRow(key, new_val, &pending);
break;
@@ -134,6 +134,9 @@ class TestRandomAccess : public KuduTabletTest {
case 4:
DeleteIgnoreRow(key, &pending); // won't change current value
break;
+ case 5:
+ cur_val = UpsertIgnoreRow(key, new_val, cur_val, &pending);
+ break;
default:
cur_val = UpsertRow(key, new_val, cur_val, &pending);
}
@@ -148,7 +151,7 @@ class TestRandomAccess : public KuduTabletTest {
} else {
// If row already exists, randomly choose between an update,
// update ignore, insert ignore, and upsert.
- switch (rand() % 4) {
+ switch (rand() % 5) {
case 1:
cur_val = UpdateRow(key, new_val, cur_val, &pending);
break;
@@ -158,6 +161,9 @@ class TestRandomAccess : public KuduTabletTest {
case 3:
InsertIgnoreRow(key, new_val, &pending); // won't change current value
break;
+ case 4:
+ cur_val = UpsertIgnoreRow(key, new_val, cur_val, &pending);
+ break;
default:
cur_val = UpsertRow(key, new_val, cur_val, &pending);
}
@@ -227,6 +233,13 @@ class TestRandomAccess : public KuduTabletTest {
return DoRowOp(RowOperationsPB::UPSERT, key, val, old_row, ops);
}
+ optional<ExpectedKeyValueRow> UpsertIgnoreRow(int key,
+ int val,
+ const optional<ExpectedKeyValueRow>& old_row,
+ vector<LocalTabletWriter::RowOp>* ops) {
+ return DoRowOp(RowOperationsPB::UPSERT_IGNORE, key, val, old_row, ops);
+ }
+
// Adds an update of the given key/value pair to 'ops', returning the expected value
optional<ExpectedKeyValueRow> UpdateRow(int key,
uint32_t new_val,
@@ -256,6 +269,7 @@ class TestRandomAccess : public KuduTabletTest {
switch (type) {
case RowOperationsPB::UPSERT:
+ case RowOperationsPB::UPSERT_IGNORE:
case RowOperationsPB::UPDATE:
case RowOperationsPB::UPDATE_IGNORE:
case RowOperationsPB::INSERT:
diff --git a/src/kudu/tablet/txn_participant-test.cc b/src/kudu/tablet/txn_participant-test.cc
index d4b27df2d..1413ebde1 100644
--- a/src/kudu/tablet/txn_participant-test.cc
+++ b/src/kudu/tablet/txn_participant-test.cc
@@ -1085,6 +1085,8 @@ TEST_F(TxnParticipantTest, TestUnsupportedOps) {
ASSERT_OK(CallParticipantOpCheckResp(kTxnId, ParticipantOpPB::BEGIN_TXN, -1));
Status s = Write(0, kTxnId, RowOperationsPB::UPSERT);
ASSERT_TRUE(s.IsNotSupported()) << s.ToString();
+ s = Write(0, kTxnId, RowOperationsPB::UPSERT_IGNORE);
+ ASSERT_TRUE(s.IsNotSupported()) << s.ToString();
s = Write(0, kTxnId, RowOperationsPB::UPDATE);
ASSERT_TRUE(s.IsNotSupported()) << s.ToString();
s = Write(0, kTxnId, RowOperationsPB::UPDATE_IGNORE);
diff --git a/src/kudu/tools/kudu-tool-test.cc b/src/kudu/tools/kudu-tool-test.cc
index e568eecff..e636dcd36 100644
--- a/src/kudu/tools/kudu-tool-test.cc
+++ b/src/kudu/tools/kudu-tool-test.cc
@@ -67,7 +67,6 @@
#include "kudu/consensus/log_util.h"
#include "kudu/consensus/opid.pb.h"
#include "kudu/consensus/opid_util.h"
-#include "kudu/consensus/raft_consensus.h"
#include "kudu/consensus/ref_counted_replicate.h"
#include "kudu/fs/block_id.h"
#include "kudu/fs/block_manager.h"
@@ -3022,14 +3021,14 @@ void ToolTest::RunLoadgen(int num_tservers,
ColumnSchema("int64_val", INT64),
ColumnSchema("float_val", FLOAT),
ColumnSchema("double_val", DOUBLE),
- ColumnSchema("decimal32_val", DECIMAL32, false,
- NULL, NULL, ColumnStorageAttributes(),
+ ColumnSchema("decimal32_val", DECIMAL32, false, false,
+ nullptr, nullptr, ColumnStorageAttributes(),
ColumnTypeAttributes(9, 9)),
- ColumnSchema("decimal64_val", DECIMAL64, false,
- NULL, NULL, ColumnStorageAttributes(),
+ ColumnSchema("decimal64_val", DECIMAL64, false, false,
+ nullptr, nullptr, ColumnStorageAttributes(),
ColumnTypeAttributes(18, 2)),
- ColumnSchema("decimal128_val", DECIMAL128, false,
- NULL, NULL, ColumnStorageAttributes(),
+ ColumnSchema("decimal128_val", DECIMAL128, false, false,
+ nullptr, nullptr, ColumnStorageAttributes(),
ColumnTypeAttributes(38, 0)),
ColumnSchema("unixtime_micros_val", UNIXTIME_MICROS),
ColumnSchema("string_val", STRING),
diff --git a/src/kudu/tserver/tablet_server_authorization-test.cc b/src/kudu/tserver/tablet_server_authorization-test.cc
index e65dcaf36..ea7fd6a67 100644
--- a/src/kudu/tserver/tablet_server_authorization-test.cc
+++ b/src/kudu/tserver/tablet_server_authorization-test.cc
@@ -25,13 +25,13 @@
#include <set>
#include <string>
#include <tuple>
+#include <type_traits>
#include <unordered_map>
#include <unordered_set>
#include <vector>
#include <gflags/gflags.h>
#include <glog/logging.h>
-#include <google/protobuf/stubs/port.h>
#include <gtest/gtest.h>
#include "kudu/common/common.pb.h"
@@ -63,6 +63,7 @@
#include "kudu/tserver/tserver.pb.h"
#include "kudu/tserver/tserver_service.pb.h"
#include "kudu/tserver/tserver_service.proxy.h"
+#include "kudu/util/bitset.h"
#include "kudu/util/memory/arena.h"
#include "kudu/util/monotime.h"
#include "kudu/util/openssl_util.h"
@@ -474,8 +475,9 @@ string GenerateEncodedKey(int32_t val, const Schema& schema) {
// Returns a column schema PB that matches 'col', but has a different name.
void MisnamedColumnSchemaToPB(const ColumnSchema& col, ColumnSchemaPB* pb) {
ColumnSchemaToPB(ColumnSchema(kDummyColumn, col.type_info()->physical_type(), col.is_nullable(),
- col.read_default_value(), col.write_default_value(), col.attributes(),
- col.type_attributes()), pb);
+ col.is_immutable(), col.read_default_value(),
+ col.write_default_value(), col.attributes(),
+ col.type_attributes()), pb);
}
} // anonymous namespace
@@ -593,6 +595,7 @@ class ScanPrivilegeAuthzTest : public AuthzTabletServerTestBase,
auto* projected_column = pb.add_projected_columns();
bool default_bool = false;
ColumnSchemaToPB(ColumnSchema("is_deleted", DataType::IS_DELETED, /*is_nullable=*/false,
+ /*is_immutable=*/false,
/*read_default=*/&default_bool, nullptr), projected_column);
}
CHECK_OK(GenerateScanAuthzToken(privilege, pb.mutable_authz_token()));
diff --git a/src/kudu/tserver/tserver.proto b/src/kudu/tserver/tserver.proto
index a6145b63b..8f6b7f0e6 100644
--- a/src/kudu/tserver/tserver.proto
+++ b/src/kudu/tserver/tserver.proto
@@ -428,6 +428,8 @@ message ResourceMetricsPB {
optional int64 delete_ignore_errors = 14;
// Total observed commit wait duration in microseconds.
optional int64 commit_wait_duration_usec = 15;
+ // Total number of UPSERT_IGNORE operations with error.
+ optional int64 upsert_ignore_errors = 16;
}
message ScanResponsePB {
diff --git a/src/kudu/util/status.cc b/src/kudu/util/status.cc
index 3c34de27b..da1746117 100644
--- a/src/kudu/util/status.cc
+++ b/src/kudu/util/status.cc
@@ -106,6 +106,9 @@ std::string Status::CodeAsString() const {
case kEndOfFile:
type = "End of file";
break;
+ case kImmutable:
+ type = "Immutable";
+ break;
default:
LOG(FATAL) << "unreachable";
}
diff --git a/src/kudu/util/status.h b/src/kudu/util/status.h
index fe9edd967..5acf59045 100644
--- a/src/kudu/util/status.h
+++ b/src/kudu/util/status.h
@@ -311,6 +311,10 @@ class KUDU_EXPORT Status {
int64_t posix_code = -1) {
return Status(kEndOfFile, msg, msg2, posix_code);
}
+ static Status Immutable(const Slice& msg, const Slice& msg2 = Slice(),
+ int64_t posix_code = -1) {
+ return Status(kImmutable, msg, msg2, posix_code);
+ }
///@}
/// @return @c true iff the status indicates success.
@@ -370,6 +374,9 @@ class KUDU_EXPORT Status {
/// @return @c true iff the status indicates end of file.
bool IsEndOfFile() const { return code() == kEndOfFile; }
+ /// @return @c true iff the status indicates immutable.
+ bool IsImmutable() const { return code() == kImmutable; }
+
/// @return @c true iff the status indicates a disk failure.
bool IsDiskFailure() const {
switch (posix_code()) {
@@ -461,6 +468,9 @@ class KUDU_EXPORT Status {
kConfigurationError = 16,
kIncomplete = 17,
kEndOfFile = 18,
+ // kCancelled stems from AppStatusPB, although it seems nobody use it now, we still reserve it.
+ // kCancelled = 19,
+ kImmutable = 20,
// NOTE: Remember to duplicate these constants into wire_protocol.proto and
// and to add StatusTo/FromPB ser/deser cases in wire_protocol.cc !
// Also remember to make the same changes to the java client in Status.java.