This is an automated email from the ASF dual-hosted git repository.
jt2594838 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new 5ce27404e87 Filter all-null FIELD columns before tablet insert in
Java/C++/Python… (#18533)
5ce27404e87 is described below
commit 5ce27404e872ca04ae5057fe0d1dd9f1fc3e6c6a
Author: Hongzhi Gao <[email protected]>
AuthorDate: Mon Aug 31 10:34:35 2026 +0800
Filter all-null FIELD columns before tablet insert in Java/C++/Python…
(#18533)
* Filter all-null FIELD columns before tablet insert in Java/C++/Python
clients
Skip or shrink tablets that only contain null FIELD values so insert paths
avoid shipping unused measurement columns across languages.
* Keep TAG/ATTRIBUTE columns when all FIELD values are null in table model
Table-model tablet inserts may carry only time and tag/attribute columns.
Do not treat all-null FIELD columns as an empty tablet when non-FIELD
columns remain, and clarify buildInsertTabletReq skip semantics.
* Refactor C++ Tablet init and use metadata-only factory in
filterNullColumns.
Add a private constructor and createWithoutValueColumns to build tablets
without pre-allocating value columns, fix bitmap null-check bounds, and update
the Java table-model test constructor.
* Check only active tablet rows when filtering all-null columns.
BitMap is sized to maxRowNumber, so a full-map isAllMarked() can miss
all-null FIELD columns when rowSize is smaller. Use isRangeAllMarked on
[0, rowSize) in the Java, C++, and Python clients.
* Keep null measurement names when filtering all-null tablet columns.
A never-written column is treated as all-null; dropping it skipped the
existing measurement-name check and omitted timeseries that tests expect.
---
.../iotdb/session/it/IoTDBSessionSimpleIT.java | 5 +-
iotdb-client/client-cpp/src/include/Common.h | 1 +
iotdb-client/client-cpp/src/include/Session.h | 50 +++---
iotdb-client/client-cpp/src/rpc/SessionImpl.h | 4 +-
iotdb-client/client-cpp/src/session/Common.cpp | 38 ++++-
iotdb-client/client-cpp/src/session/Session.cpp | 181 +++++++++++++++++----
iotdb-client/client-cpp/test/CMakeLists.txt | 7 +-
.../client-cpp/test/cpp/sessionUtilsTest.cpp | 145 +++++++++++++++++
iotdb-client/client-cpp/test/main_utils.cpp | 22 +++
iotdb-client/client-py/iotdb/Session.py | 102 ++++++++----
iotdb-client/client-py/iotdb/utils/BitMap.py | 39 +++++
iotdb-client/client-py/iotdb/utils/SessionUtils.py | 128 +++++++++++++++
iotdb-client/client-py/iotdb/utils/Tablet.py | 6 +
.../client-py/tests/unit/test_session_utils.py | 160 ++++++++++++++++++
.../java/org/apache/iotdb/session/Session.java | 114 +++++++++++--
.../apache/iotdb/session/util/SessionUtils.java | 86 ++++++++++
.../iotdb/session/util/SessionUtilsTest.java | 83 ++++++++++
17 files changed, 1053 insertions(+), 118 deletions(-)
diff --git
a/integration-test/src/test/java/org/apache/iotdb/session/it/IoTDBSessionSimpleIT.java
b/integration-test/src/test/java/org/apache/iotdb/session/it/IoTDBSessionSimpleIT.java
index 38f6345bae5..79a2225ca3b 100644
---
a/integration-test/src/test/java/org/apache/iotdb/session/it/IoTDBSessionSimpleIT.java
+++
b/integration-test/src/test/java/org/apache/iotdb/session/it/IoTDBSessionSimpleIT.java
@@ -2164,6 +2164,7 @@ public class IoTDBSessionSimpleIT {
tablet.addValue("s8", 0, new Binary(new byte[] {1}));
tablet.addValue("s9", 0, "string_value");
tablet.addValue("s10", 0, DateUtils.parseIntToLocalDate(20250403));
+ tablet.addValue("s11", 0, 1L);
try (ISession session = EnvFactory.getEnv().getSessionConnection()) {
session.insertTablet(tablet);
@@ -2193,8 +2194,8 @@ public class IoTDBSessionSimpleIT {
assertEquals("string_value", iterator.getString("root.sg.d1.s9"));
assertFalse(iterator.isNull("root.sg.d1.s10"));
assertEquals(DateUtils.parseIntToLocalDate(20250403),
iterator.getDate("root.sg.d1.s10"));
- assertTrue(iterator.isNull("root.sg.d1.s11"));
- assertNull(iterator.getTimestamp("root.sg.d1.s11"));
+ assertFalse(iterator.isNull("root.sg.d1.s11"));
+ assertEquals(new Timestamp(1),
iterator.getTimestamp("root.sg.d1.s11"));
assertEquals(new Timestamp(0), iterator.getTimestamp("Time"));
assertFalse(iterator.isNull("Time"));
diff --git a/iotdb-client/client-cpp/src/include/Common.h
b/iotdb-client/client-cpp/src/include/Common.h
index f36b38206b8..7a63c3f9341 100644
--- a/iotdb-client/client-cpp/src/include/Common.h
+++ b/iotdb-client/client-cpp/src/include/Common.h
@@ -246,6 +246,7 @@ public:
bool isMarked(size_t position) const;
bool isAllUnmarked() const;
bool isAllMarked() const;
+ bool isRangeAllMarked(size_t start, size_t length) const;
const std::vector<char>& getByteArray() const;
size_t getSize() const;
diff --git a/iotdb-client/client-cpp/src/include/Session.h
b/iotdb-client/client-cpp/src/include/Session.h
index a0910584e57..76895a30291 100644
--- a/iotdb-client/client-cpp/src/include/Session.h
+++ b/iotdb-client/client-cpp/src/include/Session.h
@@ -99,12 +99,21 @@ template <typename T, typename Target> void safe_cast(const
T& value, Target& ta
*
*/
class Tablet {
+ friend class SessionUtils;
+
private:
static const int DEFAULT_ROW_SIZE = 1024;
void createColumns();
void deleteColumns();
+ struct WithoutValueColumnsTag {};
+
+ Tablet(const std::string& deviceId,
+ const std::vector<std::pair<std::string, TSDataType::TSDataType>>&
schemas,
+ const std::vector<ColumnCategory> columnTypes, size_t maxRowNumber,
bool isAligned,
+ WithoutValueColumnsTag);
+
public:
std::string deviceId; // deviceId of this tablet
std::vector<std::pair<std::string, TSDataType::TSDataType>> schemas;
@@ -160,31 +169,18 @@ public:
const std::vector<std::pair<std::string, TSDataType::TSDataType>>&
schemas,
const std::vector<ColumnCategory> columnTypes, size_t maxRowNumber,
bool _isAligned = false)
- : deviceId(deviceId), schemas(schemas), columnTypes(columnTypes),
maxRowNumber(maxRowNumber),
- isAligned(_isAligned) {
- // create timestamp column
- timestamps.resize(maxRowNumber);
- // create value columns
- values.resize(schemas.size());
+ : Tablet(deviceId, schemas, columnTypes, maxRowNumber, _isAligned,
WithoutValueColumnsTag{}) {
createColumns();
- // init tagColumnIndexes
- for (size_t i = 0; i < this->columnTypes.size(); i++) {
- if (this->columnTypes[i] == ColumnCategory::TAG) {
- tagColumnIndexes.push_back(i);
- }
- }
- // create bitMaps
- bitMaps.resize(schemas.size());
- for (size_t i = 0; i < schemas.size(); i++) {
- bitMaps[i].resize(maxRowNumber);
- }
- // create schemaNameIndex
- for (size_t i = 0; i < schemas.size(); i++) {
- schemaNameIndex[schemas[i].first] = i;
- }
- this->rowSize = 0;
}
+ /**
+ * Create a tablet with metadata and bitmaps only; value columns are filled
by the caller.
+ */
+ static std::shared_ptr<Tablet> createWithoutValueColumns(
+ const std::string& deviceId,
+ const std::vector<std::pair<std::string, TSDataType::TSDataType>>&
schemas,
+ const std::vector<ColumnCategory>& columnTypes, size_t maxRowNumber,
bool isAligned = false);
+
Tablet(const Tablet& other)
: deviceId(other.deviceId), schemas(other.schemas),
schemaNameIndex(other.schemaNameIndex),
columnTypes(other.columnTypes), timestamps(other.timestamps),
@@ -431,6 +427,16 @@ public:
static std::string getValue(const Tablet& tablet);
static bool isTabletContainsSingleDevice(Tablet tablet);
+
+ /**
+ * Drop entirely-null FIELD columns within [0, rowSize). TAG/ATTRIBUTE are
kept.
+ * Does not mutate {@code tablet}.
+ *
+ * @return non-owning pointer to {@code tablet} if nothing to drop; a new
tablet if filtered;
+ * nullptr if no columns remain. Table-model TAG / ATTRIBUTE columns
are kept even when
+ * every FIELD column is null.
+ */
+ static std::shared_ptr<const Tablet> filterNullColumns(const Tablet& tablet);
};
class TemplateNode {
diff --git a/iotdb-client/client-cpp/src/rpc/SessionImpl.h
b/iotdb-client/client-cpp/src/rpc/SessionImpl.h
index 9fc3d917293..406537486c0 100644
--- a/iotdb-client/client-cpp/src/rpc/SessionImpl.h
+++ b/iotdb-client/client-cpp/src/rpc/SessionImpl.h
@@ -145,7 +145,9 @@ public:
void handleRedirection(const std::string& deviceId, TEndPoint endPoint);
void handleRedirection(const std::shared_ptr<storage::IDeviceID>& deviceId,
TEndPoint endPoint);
- static void buildInsertTabletReq(TSInsertTabletReq& request, Tablet& tablet,
bool sorted);
+ // Returns false when filtering leaves no columns (tree-model all-null FIELD
tablet).
+ // Table-model inserts may keep TAG / ATTRIBUTE columns when all FIELD
columns are null.
+ static bool buildInsertTabletReq(TSInsertTabletReq& request, Tablet& tablet,
bool sorted);
void insertTablet(TSInsertTabletReq request);
void insertRelationalTabletOnce(
const std::unordered_map<std::shared_ptr<SessionConnection>, Tablet>&
relationalTabletGroup,
diff --git a/iotdb-client/client-cpp/src/session/Common.cpp
b/iotdb-client/client-cpp/src/session/Common.cpp
index 859a8209a11..3dd829b95b0 100644
--- a/iotdb-client/client-cpp/src/session/Common.cpp
+++ b/iotdb-client/client-cpp/src/session/Common.cpp
@@ -323,18 +323,40 @@ bool BitMap::isAllUnmarked() const {
}
bool BitMap::isAllMarked() const {
- size_t j;
- for (j = 0; j < size >> 3; j++) {
- if (bits[j] != (char)0XFF) {
- return false;
- }
+ return isRangeAllMarked(0, size);
+}
+
+bool BitMap::isRangeAllMarked(size_t start, size_t length) const {
+ // size_t addition can wrap; short-circuit so `size - start` is only used
when
+ // start <= size. Out-of-range matches mark()/isMarked(): return false.
+ if (start > size || length > size - start) {
+ return false;
}
- for (j = 0; j < size % 8; j++) {
- if ((bits[size >> 3] & ((char)1 << j)) == 0) {
+ if (length == 0) {
+ return true;
+ }
+
+ const size_t end = start + length;
+ const size_t firstByte = start >> 3;
+ const size_t lastByte = (end - 1) >> 3;
+ if (firstByte == lastByte) {
+ const unsigned char mask = static_cast<unsigned char>(((1U << length) -
1U) << (start & 7U));
+ return (static_cast<unsigned char>(bits[firstByte]) & mask) == mask;
+ }
+
+ const unsigned char firstMask = static_cast<unsigned char>((0xFFU << (start
& 7U)) & 0xFFU);
+ if ((static_cast<unsigned char>(bits[firstByte]) & firstMask) != firstMask) {
+ return false;
+ }
+ for (size_t index = firstByte + 1; index < lastByte; index++) {
+ if (static_cast<unsigned char>(bits[index]) != 0xFFU) {
return false;
}
}
- return true;
+ const size_t lastBitCount = end & 7U;
+ const unsigned char lastMask =
+ lastBitCount == 0 ? 0xFFU : static_cast<unsigned char>((1U <<
lastBitCount) - 1U);
+ return (static_cast<unsigned char>(bits[lastByte]) & lastMask) == lastMask;
}
const std::vector<char>& BitMap::getByteArray() const {
diff --git a/iotdb-client/client-cpp/src/session/Session.cpp
b/iotdb-client/client-cpp/src/session/Session.cpp
index cf429af0c4d..ab71732ebe1 100644
--- a/iotdb-client/client-cpp/src/session/Session.cpp
+++ b/iotdb-client/client-cpp/src/session/Session.cpp
@@ -69,6 +69,39 @@ TSDataType::TSDataType getTSDataTypeFromString(const string&
str) {
return TSDataType::UNKNOWN;
}
+Tablet::Tablet(const std::string& deviceId,
+ const std::vector<std::pair<std::string,
TSDataType::TSDataType>>& schemas,
+ const std::vector<ColumnCategory> columnTypes, size_t
maxRowNumber, bool isAligned,
+ WithoutValueColumnsTag)
+ : deviceId(deviceId), schemas(schemas), columnTypes(columnTypes),
maxRowNumber(maxRowNumber),
+ isAligned(isAligned) {
+ timestamps.resize(maxRowNumber);
+ values.resize(schemas.size(), nullptr);
+ tagColumnIndexes.clear();
+ for (size_t i = 0; i < columnTypes.size(); i++) {
+ if (columnTypes[i] == ColumnCategory::TAG) {
+ tagColumnIndexes.push_back(static_cast<int>(i));
+ }
+ }
+ bitMaps.resize(schemas.size());
+ for (size_t i = 0; i < schemas.size(); i++) {
+ bitMaps[i].resize(maxRowNumber);
+ }
+ schemaNameIndex.clear();
+ for (size_t i = 0; i < schemas.size(); i++) {
+ schemaNameIndex[schemas[i].first] = i;
+ }
+ rowSize = 0;
+}
+
+std::shared_ptr<Tablet> Tablet::createWithoutValueColumns(
+ const std::string& deviceId,
+ const std::vector<std::pair<std::string, TSDataType::TSDataType>>& schemas,
+ const std::vector<ColumnCategory>& columnTypes, size_t maxRowNumber, bool
isAligned) {
+ return std::shared_ptr<Tablet>(new Tablet(deviceId, schemas, columnTypes,
maxRowNumber, isAligned,
+ WithoutValueColumnsTag{}));
+}
+
void Tablet::createColumns() {
for (size_t i = 0; i < schemas.size(); i++) {
TSDataType::TSDataType dataType = schemas[i].second;
@@ -413,6 +446,64 @@ bool SessionUtils::isTabletContainsSingleDevice(Tablet
tablet) {
return true;
}
+static bool isColumnAllNull(const BitMap& bitMap, size_t rowSize) {
+ if (rowSize == 0) {
+ return false;
+ }
+ // BitMap is sized to maxRowNumber; only [0, rowSize) are active rows.
+ return bitMap.isRangeAllMarked(0, rowSize);
+}
+
+std::shared_ptr<const Tablet> SessionUtils::filterNullColumns(const Tablet&
tablet) {
+ const size_t columnCount = tablet.schemas.size();
+ if (columnCount == 0 || tablet.bitMaps.size() < columnCount) {
+ return std::shared_ptr<const Tablet>(&tablet, [](const Tablet*) {});
+ }
+
+ std::vector<size_t> keptIndices;
+ keptIndices.reserve(columnCount);
+
+ for (size_t i = 0; i < columnCount; i++) {
+ ColumnCategory category =
+ i < tablet.columnTypes.size() ? tablet.columnTypes[i] :
ColumnCategory::FIELD;
+ bool isField = category == ColumnCategory::FIELD;
+ bool drop = isField && isColumnAllNull(tablet.bitMaps[i], tablet.rowSize);
+ if (drop) {
+ continue;
+ }
+ keptIndices.push_back(i);
+ }
+
+ if (keptIndices.size() == columnCount) {
+ return std::shared_ptr<const Tablet>(&tablet, [](const Tablet*) {});
+ }
+ if (keptIndices.empty()) {
+ return nullptr;
+ }
+
+ std::vector<std::pair<std::string, TSDataType::TSDataType>> keptSchemas;
+ std::vector<ColumnCategory> keptColumnTypes;
+ keptSchemas.reserve(keptIndices.size());
+ keptColumnTypes.reserve(keptIndices.size());
+ for (size_t idx : keptIndices) {
+ keptSchemas.push_back(tablet.schemas[idx]);
+ keptColumnTypes.push_back(idx < tablet.columnTypes.size() ?
tablet.columnTypes[idx]
+ :
ColumnCategory::FIELD);
+ }
+
+ auto filteredOut = Tablet::createWithoutValueColumns(
+ tablet.deviceId, keptSchemas, keptColumnTypes, tablet.maxRowNumber,
tablet.isAligned);
+ filteredOut->timestamps = tablet.timestamps;
+ filteredOut->rowSize = tablet.rowSize;
+ for (size_t ni = 0; ni < keptIndices.size(); ni++) {
+ size_t oi = keptIndices[ni];
+ Tablet::deepCopyTabletColValue(&tablet.values[oi],
&filteredOut->values[ni],
+ keptSchemas[ni].second,
static_cast<int>(tablet.maxRowNumber));
+ filteredOut->bitMaps[ni] = tablet.bitMaps[oi];
+ }
+ return filteredOut;
+}
+
string MeasurementNode::serialize() const {
MyStringBuffer buffer;
buffer.putString(getName());
@@ -1002,6 +1093,10 @@ void
Session::Impl::insertTabletsWithLeaderCache(unordered_map<string, Tablet*>&
}
auto deviceId = item.first;
auto tablet = item.second;
+ std::shared_ptr<const Tablet> toEncode =
SessionUtils::filterNullColumns(*tablet);
+ if (!toEncode) {
+ continue;
+ }
auto connection = getSessionConnection(deviceId);
auto it = tabletsGroup.find(connection);
if (it == tabletsGroup.end()) {
@@ -1009,13 +1104,13 @@ void
Session::Impl::insertTabletsWithLeaderCache(unordered_map<string, Tablet*>&
tabletsGroup[connection] = request;
}
TSInsertTabletsReq& existingReq = tabletsGroup[connection];
- existingReq.prefixPaths.emplace_back(tablet->deviceId);
-
existingReq.timestampsList.emplace_back(move(SessionUtils::getTime(*tablet)));
- existingReq.valuesList.emplace_back(move(SessionUtils::getValue(*tablet)));
- existingReq.sizeList.emplace_back(tablet->rowSize);
+ existingReq.prefixPaths.emplace_back(toEncode->deviceId);
+
existingReq.timestampsList.emplace_back(move(SessionUtils::getTime(*toEncode)));
+
existingReq.valuesList.emplace_back(move(SessionUtils::getValue(*toEncode)));
+ existingReq.sizeList.emplace_back(toEncode->rowSize);
vector<int> dataTypes;
vector<string> measurements;
- for (pair<string, TSDataType::TSDataType> schema : tablet->schemas) {
+ for (pair<string, TSDataType::TSDataType> schema : toEncode->schemas) {
measurements.push_back(schema.first);
dataTypes.push_back(schema.second);
}
@@ -1023,6 +1118,10 @@ void
Session::Impl::insertTabletsWithLeaderCache(unordered_map<string, Tablet*>&
existingReq.typesList.emplace_back(dataTypes);
}
+ if (tabletsGroup.empty()) {
+ return;
+ }
+
std::function<void(std::shared_ptr<SessionConnection>, const
TSInsertTabletsReq&)> consumer =
[](const std::shared_ptr<SessionConnection>& c, const
TSInsertTabletsReq& r) {
c->insertTablets(r);
@@ -1444,27 +1543,41 @@ void Session::insertTablet(Tablet& tablet) {
}
}
-void Session::Impl::buildInsertTabletReq(TSInsertTabletReq& request, Tablet&
tablet, bool sorted) {
+bool Session::Impl::buildInsertTabletReq(TSInsertTabletReq& request, Tablet&
tablet, bool sorted) {
if ((!sorted) && !checkSorted(tablet)) {
sortTablet(tablet);
}
- request.__set_prefixPath(tablet.deviceId);
+ std::shared_ptr<const Tablet> toEncode =
SessionUtils::filterNullColumns(tablet);
+ if (!toEncode) {
+ return false;
+ }
+
+ request.__set_prefixPath(toEncode->deviceId);
std::vector<std::string> reqMeasurements;
- reqMeasurements.reserve(tablet.schemas.size());
+ reqMeasurements.reserve(toEncode->schemas.size());
std::vector<int32_t> types;
- types.reserve(tablet.schemas.size());
- for (pair<string, TSDataType::TSDataType> schema : tablet.schemas) {
+ types.reserve(toEncode->schemas.size());
+ for (pair<string, TSDataType::TSDataType> schema : toEncode->schemas) {
reqMeasurements.push_back(schema.first);
types.push_back(schema.second);
}
request.__set_measurements(reqMeasurements);
request.__set_types(types);
- request.__set_values(SessionUtils::getValue(tablet));
- request.__set_timestamps(SessionUtils::getTime(tablet));
- request.__set_size(tablet.rowSize);
- request.__set_isAligned(tablet.isAligned);
+ request.__set_values(SessionUtils::getValue(*toEncode));
+ request.__set_timestamps(SessionUtils::getTime(*toEncode));
+ request.__set_size(toEncode->rowSize);
+ request.__set_isAligned(toEncode->isAligned);
+ if (!toEncode->columnTypes.empty()) {
+ std::vector<int8_t> columnCategories;
+ columnCategories.reserve(toEncode->columnTypes.size());
+ for (auto& category : toEncode->columnTypes) {
+ columnCategories.push_back(static_cast<int8_t>(category));
+ }
+ request.__set_columnCategories(columnCategories);
+ }
+ return true;
}
void Session::Impl::insertTablet(TSInsertTabletReq request) {
@@ -1488,7 +1601,9 @@ void Session::Impl::insertTablet(TSInsertTabletReq
request) {
void Session::insertTablet(Tablet& tablet, bool sorted) {
TSInsertTabletReq request;
- impl_->buildInsertTabletReq(request, tablet, sorted);
+ if (!impl_->buildInsertTabletReq(request, tablet, sorted)) {
+ return;
+ }
impl_->insertTablet(request);
}
@@ -1574,13 +1689,10 @@ void Session::Impl::insertRelationalTabletOnce(
auto connection = iter->first;
auto tablet = iter->second;
TSInsertTabletReq request;
- buildInsertTabletReq(request, tablet, sorted);
- request.__set_writeToTable(true);
- std::vector<int8_t> columnCategories;
- for (auto& category : tablet.columnTypes) {
- columnCategories.push_back(static_cast<int8_t>(category));
+ if (!buildInsertTabletReq(request, tablet, sorted)) {
+ return;
}
- request.__set_columnCategories(columnCategories);
+ request.__set_writeToTable(true);
try {
TSStatus respStatus;
connection->getSessionClient()->insertTablet(respStatus, request);
@@ -1629,14 +1741,10 @@ void Session::Impl::insertRelationalTabletByGroup(
futures.emplace_back(
std::async(std::launch::async, [this, connection, tablet, sorted]()
mutable {
TSInsertTabletReq request;
- buildInsertTabletReq(request, tablet, sorted);
- request.__set_writeToTable(true);
-
- std::vector<int8_t> columnCategories;
- for (auto& category : tablet.columnTypes) {
- columnCategories.push_back(static_cast<int8_t>(category));
+ if (!buildInsertTabletReq(request, tablet, sorted)) {
+ return;
}
- request.__set_columnCategories(columnCategories);
+ request.__set_writeToTable(true);
try {
TSStatus respStatus;
@@ -1702,18 +1810,25 @@ void Session::insertTablets(unordered_map<string,
Tablet*>& tablets, bool sorted
if (!impl_->checkSorted(*(item.second))) {
impl_->sortTablet(*(item.second));
}
- request.prefixPaths.push_back(item.second->deviceId);
+ std::shared_ptr<const Tablet> toEncode =
SessionUtils::filterNullColumns(*(item.second));
+ if (!toEncode) {
+ continue;
+ }
+ request.prefixPaths.push_back(toEncode->deviceId);
vector<string> measurements;
vector<int> dataTypes;
- for (pair<string, TSDataType::TSDataType> schema : item.second->schemas)
{
+ for (pair<string, TSDataType::TSDataType> schema : toEncode->schemas) {
measurements.push_back(schema.first);
dataTypes.push_back(schema.second);
}
request.measurementsList.push_back(measurements);
request.typesList.push_back(dataTypes);
-
request.timestampsList.push_back(move(SessionUtils::getTime(*(item.second))));
-
request.valuesList.push_back(move(SessionUtils::getValue(*(item.second))));
- request.sizeList.push_back(item.second->rowSize);
+ request.timestampsList.push_back(move(SessionUtils::getTime(*toEncode)));
+ request.valuesList.push_back(move(SessionUtils::getValue(*toEncode)));
+ request.sizeList.push_back(toEncode->rowSize);
+ }
+ if (request.prefixPaths.empty()) {
+ return;
}
request.__set_isAligned(isAligned);
try {
diff --git a/iotdb-client/client-cpp/test/CMakeLists.txt
b/iotdb-client/client-cpp/test/CMakeLists.txt
index 9d5428edc4b..b4a7cf1f767 100644
--- a/iotdb-client/client-cpp/test/CMakeLists.txt
+++ b/iotdb-client/client-cpp/test/CMakeLists.txt
@@ -42,12 +42,14 @@ set(_test_targets
session_tests
session_relational_tests
session_c_tests
- session_c_relational_tests)
+ session_c_relational_tests
+ session_utils_tests)
add_executable(session_tests main.cpp
cpp/sessionIT.cpp)
add_executable(session_relational_tests main_Relational.cpp
cpp/sessionRelationalIT.cpp)
add_executable(session_c_tests main_c.cpp
cpp/sessionCIT.cpp)
add_executable(session_c_relational_tests main_c_Relational.cpp
cpp/sessionCRelationalIT.cpp)
+add_executable(session_utils_tests main_utils.cpp
cpp/sessionUtilsTest.cpp)
foreach(_t IN LISTS _test_targets)
target_include_directories(${_t} PRIVATE
@@ -86,6 +88,7 @@ if(MSVC)
add_test(NAME sessionRelationalIT CONFIGURATIONS Release COMMAND
session_relational_tests)
add_test(NAME sessionCIT CONFIGURATIONS Release COMMAND
session_c_tests)
add_test(NAME sessionCRelationalIT CONFIGURATIONS Release COMMAND
session_c_relational_tests)
+ add_test(NAME sessionUtilsTest CONFIGURATIONS Release COMMAND
session_utils_tests)
foreach(_t IN LISTS _test_targets)
add_custom_command(TARGET ${_t} POST_BUILD
COMMAND ${CMAKE_COMMAND} -E copy_if_different
@@ -96,9 +99,11 @@ else()
add_test(NAME sessionRelationalIT COMMAND session_relational_tests)
add_test(NAME sessionCIT COMMAND session_c_tests)
add_test(NAME sessionCRelationalIT COMMAND session_c_relational_tests)
+ add_test(NAME sessionUtilsTest COMMAND session_utils_tests)
endif()
# Run sequentially: parallel ctest overloads the single local IoTDB instance.
+# sessionUtilsTest is a pure unit test and can run anytime.
set_tests_properties(
sessionIT sessionRelationalIT sessionCIT sessionCRelationalIT
PROPERTIES RUN_SERIAL TRUE)
diff --git a/iotdb-client/client-cpp/test/cpp/sessionUtilsTest.cpp
b/iotdb-client/client-cpp/test/cpp/sessionUtilsTest.cpp
new file mode 100644
index 00000000000..18047a38868
--- /dev/null
+++ b/iotdb-client/client-cpp/test/cpp/sessionUtilsTest.cpp
@@ -0,0 +1,145 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#include "catch.hpp"
+#include "Session.h"
+
+#include <memory>
+#include <vector>
+
+using namespace std;
+
+TEST_CASE("SessionUtils filterNullColumns keeps only non-null FIELD columns",
"[utils]") {
+ vector<pair<string, TSDataType::TSDataType>> schemas = {{"s1",
TSDataType::INT32},
+ {"s2",
TSDataType::INT64},
+ {"s3",
TSDataType::FLOAT},
+ {"s4",
TSDataType::DOUBLE},
+ {"s5",
TSDataType::BOOLEAN}};
+ Tablet tablet("root.sg.d1", schemas, 1);
+ tablet.timestamps[0] = 1000L;
+ tablet.rowSize = 1;
+ tablet.addValue("s1", 0, 1);
+ tablet.addValue("s3", 0, 1.5f);
+ tablet.bitMaps[1].mark(0);
+ tablet.bitMaps[3].mark(0);
+ tablet.bitMaps[4].mark(0);
+
+ std::shared_ptr<const Tablet> filtered =
SessionUtils::filterNullColumns(tablet);
+ REQUIRE(filtered != nullptr);
+ REQUIRE(filtered.get() != &tablet);
+ REQUIRE(filtered.use_count() == 1);
+ REQUIRE(filtered->schemas.size() == 2);
+ REQUIRE(filtered->schemas[0].first == "s1");
+ REQUIRE(filtered->schemas[1].first == "s3");
+ REQUIRE(((int*)filtered->values[0])[0] == 1);
+ REQUIRE(((float*)filtered->values[1])[0] == Approx(1.5f));
+}
+
+TEST_CASE("SessionUtils filterNullColumns returns original when nothing to
drop", "[utils]") {
+ vector<pair<string, TSDataType::TSDataType>> schemas = {{"s1",
TSDataType::INT32}};
+ Tablet tablet("root.sg.d1", schemas, 1);
+ tablet.timestamps[0] = 1L;
+ tablet.rowSize = 1;
+ tablet.addValue("s1", 0, 1);
+
+ std::shared_ptr<const Tablet> filtered =
SessionUtils::filterNullColumns(tablet);
+ REQUIRE(filtered.get() == &tablet);
+}
+
+TEST_CASE("SessionUtils filterNullColumns returns nullptr when tree-model
FIELD columns are all null",
+ "[utils]") {
+ vector<pair<string, TSDataType::TSDataType>> schemas = {{"s1",
TSDataType::INT32},
+ {"s2",
TSDataType::INT64}};
+ Tablet tablet("root.sg.d1", schemas, 1);
+ tablet.timestamps[0] = 2000L;
+ tablet.rowSize = 1;
+ tablet.bitMaps[0].mark(0);
+ tablet.bitMaps[1].mark(0);
+
+ std::shared_ptr<const Tablet> filtered =
SessionUtils::filterNullColumns(tablet);
+ REQUIRE(filtered == nullptr);
+}
+
+TEST_CASE("SessionUtils filterNullColumns keeps TAG when table-model FIELD
columns are all null",
+ "[utils]") {
+ vector<pair<string, TSDataType::TSDataType>> schemas = {
+ {"tag1", TSDataType::TEXT}, {"s1", TSDataType::INT32}, {"s2",
TSDataType::INT64}};
+ vector<ColumnCategory> columnTypes = {ColumnCategory::TAG,
ColumnCategory::FIELD,
+ ColumnCategory::FIELD};
+ Tablet tablet("table1", schemas, columnTypes, 1);
+ tablet.timestamps[0] = 3000L;
+ tablet.rowSize = 1;
+ tablet.addValue("tag1", 0, string("d1"));
+ tablet.bitMaps[1].mark(0);
+ tablet.bitMaps[2].mark(0);
+
+ std::shared_ptr<const Tablet> filtered =
SessionUtils::filterNullColumns(tablet);
+ REQUIRE(filtered != nullptr);
+ REQUIRE(filtered.get() != &tablet);
+ REQUIRE(filtered->schemas.size() == 1);
+ REQUIRE(filtered->schemas[0].first == "tag1");
+ REQUIRE(filtered->columnTypes.size() == 1);
+ REQUIRE(filtered->columnTypes[0] == ColumnCategory::TAG);
+}
+
+TEST_CASE("SessionUtils filterNullColumns checks active rows when bitmap is
maxRowNumber-sized",
+ "[utils]") {
+ vector<pair<string, TSDataType::TSDataType>> schemas = {{"s1",
TSDataType::INT32},
+ {"s2",
TSDataType::INT64}};
+ Tablet tablet("root.sg.d1", schemas, 10);
+ tablet.timestamps[0] = 1000L;
+ tablet.rowSize = 1;
+ tablet.addValue("s1", 0, 1);
+ tablet.bitMaps[1].mark(0);
+
+ std::shared_ptr<const Tablet> filtered =
SessionUtils::filterNullColumns(tablet);
+ REQUIRE(filtered != nullptr);
+ REQUIRE(filtered->schemas.size() == 1);
+ REQUIRE(filtered->schemas[0].first == "s1");
+}
+
+TEST_CASE("BitMap isRangeAllMarked matches per-bit scan and rejects OOB",
"[utils]") {
+ BitMap bitMap(20);
+ for (size_t i = 0; i < 20; i++) {
+ if (i % 4 != 2) {
+ bitMap.mark(i);
+ }
+ }
+ for (size_t start = 0; start <= 20; start++) {
+ for (size_t length = 0; length <= 20 - start; length++) {
+ bool allMarked = true;
+ for (size_t i = start; i < start + length; i++) {
+ allMarked = allMarked && bitMap.isMarked(i);
+ }
+ REQUIRE(bitMap.isRangeAllMarked(start, length) == allMarked);
+ }
+ }
+ REQUIRE(bitMap.isRangeAllMarked(0, 0));
+ REQUIRE_FALSE(bitMap.isRangeAllMarked(0, 21));
+ REQUIRE_FALSE(bitMap.isRangeAllMarked(21, 0));
+
+ BitMap empty;
+ REQUIRE(empty.isAllMarked());
+ BitMap full(8);
+ full.markAll();
+ REQUIRE(full.isAllMarked());
+ full.unmark(7);
+ REQUIRE_FALSE(full.isAllMarked());
+ REQUIRE(full.isRangeAllMarked(0, 7));
+}
diff --git a/iotdb-client/client-cpp/test/main_utils.cpp
b/iotdb-client/client-cpp/test/main_utils.cpp
new file mode 100644
index 00000000000..3c9c54084cb
--- /dev/null
+++ b/iotdb-client/client-cpp/test/main_utils.cpp
@@ -0,0 +1,22 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#define CATCH_CONFIG_MAIN
+
+#include <catch.hpp>
diff --git a/iotdb-client/client-py/iotdb/Session.py
b/iotdb-client/client-py/iotdb/Session.py
index b962e3ead5e..5a4f2ff23e7 100644
--- a/iotdb-client/client-py/iotdb/Session.py
+++ b/iotdb-client/client-py/iotdb/Session.py
@@ -24,6 +24,7 @@ import sys
import warnings
from iotdb.utils.SessionDataSet import SessionDataSet
+from iotdb.utils.SessionUtils import filter_null_columns
from thrift.protocol import TBinaryProtocol, TCompactProtocol
from thrift.transport import TSocket, TTransport
from tzlocal import get_localzone_name
@@ -1095,6 +1096,8 @@ class Session(object):
:param tablet: a tablet specified above
"""
request = self.gen_insert_tablet_req(tablet)
+ if request is None:
+ return
try:
connection = self.get_connection(tablet.get_insert_target_name())
request.sessionId = connection.session_id
@@ -1123,19 +1126,24 @@ class Session(object):
if self.__enable_redirection:
request_group = {}
for i in range(len(tablet_lst)):
- connection =
self.get_connection(tablet_lst[i].get_insert_target_name())
+ filtered = filter_null_columns(tablet_lst[i])
+ if filtered is None:
+ continue
+ connection =
self.get_connection(filtered.get_insert_target_name())
request = request_group.setdefault(
connection.client,
TSInsertTabletsReq(
connection.session_id, [], [], [], [], [], [], False
),
)
-
request.prefixPaths.append(tablet_lst[i].get_insert_target_name())
-
request.timestampsList.append(tablet_lst[i].get_binary_timestamps())
-
request.measurementsList.append(tablet_lst[i].get_measurements())
- request.valuesList.append(tablet_lst[i].get_binary_values())
- request.sizeList.append(tablet_lst[i].get_row_number())
- request.typesList.append(tablet_lst[i].get_data_types())
+ request.prefixPaths.append(filtered.get_insert_target_name())
+ request.timestampsList.append(filtered.get_binary_timestamps())
+ request.measurementsList.append(filtered.get_measurements())
+ request.valuesList.append(filtered.get_binary_values())
+ request.sizeList.append(filtered.get_row_number())
+ request.typesList.append(filtered.get_data_types())
+ if not request_group:
+ return 0
for client, request in request_group.items():
try:
rpc_utils.verify_success_with_redirection_for_multi_devices(
@@ -1161,6 +1169,8 @@ class Session(object):
return 0
else:
request = self.gen_insert_tablets_req(tablet_lst)
+ if request is None:
+ return 0
try:
return
rpc_utils.verify_success(self.__client.insertTablets(request))
except TTransport.TException as e:
@@ -1190,6 +1200,8 @@ class Session(object):
:param tablet: a tablet specified above
"""
request = self.gen_insert_tablet_req(tablet, True)
+ if request is None:
+ return
try:
connection = self.get_connection(tablet.get_insert_target_name())
request.sessionId = connection.session_id
@@ -1218,19 +1230,24 @@ class Session(object):
if self.__enable_redirection:
request_group = {}
for i in range(len(tablet_lst)):
- connection =
self.get_connection(tablet_lst[i].get_insert_target_name())
+ filtered = filter_null_columns(tablet_lst[i])
+ if filtered is None:
+ continue
+ connection =
self.get_connection(filtered.get_insert_target_name())
request = request_group.setdefault(
connection.client,
TSInsertTabletsReq(
connection.session_id, [], [], [], [], [], [], True
),
)
-
request.prefixPaths.append(tablet_lst[i].get_insert_target_name())
-
request.timestampsList.append(tablet_lst[i].get_binary_timestamps())
-
request.measurementsList.append(tablet_lst[i].get_measurements())
- request.valuesList.append(tablet_lst[i].get_binary_values())
- request.sizeList.append(tablet_lst[i].get_row_number())
- request.typesList.append(tablet_lst[i].get_data_types())
+ request.prefixPaths.append(filtered.get_insert_target_name())
+ request.timestampsList.append(filtered.get_binary_timestamps())
+ request.measurementsList.append(filtered.get_measurements())
+ request.valuesList.append(filtered.get_binary_values())
+ request.sizeList.append(filtered.get_row_number())
+ request.typesList.append(filtered.get_data_types())
+ if not request_group:
+ return 0
for client, request in request_group.items():
try:
rpc_utils.verify_success_with_redirection_for_multi_devices(
@@ -1256,6 +1273,8 @@ class Session(object):
return 0
else:
request = self.gen_insert_tablets_req(tablet_lst, True)
+ if request is None:
+ return 0
try:
return
rpc_utils.verify_success(self.__client.insertTablets(request))
except TTransport.TException as e:
@@ -1282,6 +1301,8 @@ class Session(object):
:param tablet: a tablet specified above
"""
request = self.gen_insert_relational_tablet_req(tablet)
+ if request is None:
+ return
try:
connection = self.get_connection(tablet.get_insert_target_name())
request.sessionId = connection.session_id
@@ -1480,6 +1501,8 @@ class Session(object):
:param tablet: a tablet of data
"""
request = self.gen_insert_tablet_req(tablet)
+ if request is None:
+ return
try:
return
rpc_utils.verify_success(self.__client.testInsertTablet(request))
except TTransport.TException as e:
@@ -1501,6 +1524,8 @@ class Session(object):
:param tablet_list: List of tablets
"""
request = self.gen_insert_tablets_req(tablet_list)
+ if request is None:
+ return
try:
return
rpc_utils.verify_success(self.__client.testInsertTablets(request))
except TTransport.TException as e:
@@ -1516,29 +1541,35 @@ class Session(object):
raise IoTDBConnectionException(self.connection_error_msg())
from None
def gen_insert_tablet_req(self, tablet, is_aligned=False):
+ filtered = filter_null_columns(tablet)
+ if filtered is None:
+ return None
return TSInsertTabletReq(
self.__session_id,
- tablet.get_insert_target_name(),
- tablet.get_measurements(),
- tablet.get_binary_values(),
- tablet.get_binary_timestamps(),
- tablet.get_data_types(),
- tablet.get_row_number(),
+ filtered.get_insert_target_name(),
+ filtered.get_measurements(),
+ filtered.get_binary_values(),
+ filtered.get_binary_timestamps(),
+ filtered.get_data_types(),
+ filtered.get_row_number(),
is_aligned,
)
def gen_insert_relational_tablet_req(self, tablet, is_aligned=False):
+ filtered = filter_null_columns(tablet)
+ if filtered is None:
+ return None
return TSInsertTabletReq(
self.__session_id,
- tablet.get_insert_target_name(),
- tablet.get_measurements(),
- tablet.get_binary_values(),
- tablet.get_binary_timestamps(),
- tablet.get_data_types(),
- tablet.get_row_number(),
+ filtered.get_insert_target_name(),
+ filtered.get_measurements(),
+ filtered.get_binary_values(),
+ filtered.get_binary_timestamps(),
+ filtered.get_data_types(),
+ filtered.get_row_number(),
is_aligned,
True,
- tablet.get_column_categories(),
+ filtered.get_column_categories(),
)
def gen_insert_tablets_req(self, tablet_lst, is_aligned=False):
@@ -1549,12 +1580,17 @@ class Session(object):
type_lst = []
size_lst = []
for tablet in tablet_lst:
- device_id_lst.append(tablet.get_insert_target_name())
- measurements_lst.append(tablet.get_measurements())
- values_lst.append(tablet.get_binary_values())
- timestamps_lst.append(tablet.get_binary_timestamps())
- type_lst.append(tablet.get_data_types())
- size_lst.append(tablet.get_row_number())
+ filtered = filter_null_columns(tablet)
+ if filtered is None:
+ continue
+ device_id_lst.append(filtered.get_insert_target_name())
+ measurements_lst.append(filtered.get_measurements())
+ values_lst.append(filtered.get_binary_values())
+ timestamps_lst.append(filtered.get_binary_timestamps())
+ type_lst.append(filtered.get_data_types())
+ size_lst.append(filtered.get_row_number())
+ if not device_id_lst:
+ return None
return TSInsertTabletsReq(
self.__session_id,
device_id_lst,
diff --git a/iotdb-client/client-py/iotdb/utils/BitMap.py
b/iotdb-client/client-py/iotdb/utils/BitMap.py
index 621bf6c7df9..d0b57ed1383 100644
--- a/iotdb-client/client-py/iotdb/utils/BitMap.py
+++ b/iotdb-client/client-py/iotdb/utils/BitMap.py
@@ -29,6 +29,12 @@ class BitMap(object):
def mark(self, position):
self.bits[position // 8] |= BitMap.BIT_UTIL[position % 8]
+ def is_marked(self, position):
+ return (self.bits[position // 8] & BitMap.BIT_UTIL[position % 8]) != 0
+
+ def get_size(self):
+ return self.__size
+
def is_all_unmarked(self):
for i in range(self.__size // 8):
if self.bits[i] != 0:
@@ -37,3 +43,36 @@ class BitMap(object):
if (self.bits[self.__size // 8] & BitMap.BIT_UTIL[i]) != 0:
return False
return True
+
+ def is_all_marked(self):
+ return self.is_range_all_marked(0, self.__size)
+
+ def is_range_all_marked(self, start, length):
+ # Reject negatives: Python // on a negative start would index bits from
+ # the tail (bits[-1]). Out-of-range matches C++ BitMap: return False.
+ if (
+ start < 0
+ or length < 0
+ or start > self.__size
+ or length > self.__size - start
+ ):
+ return False
+ if length == 0:
+ return True
+
+ end = start + length
+ first_byte = start // 8
+ last_byte = (end - 1) // 8
+ if first_byte == last_byte:
+ mask = ((1 << length) - 1) << (start & 7)
+ return (self.bits[first_byte] & mask) == mask
+
+ first_mask = (0xFF << (start & 7)) & 0xFF
+ if (self.bits[first_byte] & first_mask) != first_mask:
+ return False
+ for index in range(first_byte + 1, last_byte):
+ if self.bits[index] != 0xFF:
+ return False
+ last_bit_count = end & 7
+ last_mask = 0xFF if last_bit_count == 0 else (1 << last_bit_count) - 1
+ return (self.bits[last_byte] & last_mask) == last_mask
diff --git a/iotdb-client/client-py/iotdb/utils/SessionUtils.py
b/iotdb-client/client-py/iotdb/utils/SessionUtils.py
new file mode 100644
index 00000000000..eaf64e6fea0
--- /dev/null
+++ b/iotdb-client/client-py/iotdb/utils/SessionUtils.py
@@ -0,0 +1,128 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+#
+
+from typing import List, Optional, Union
+
+from iotdb.utils.BitMap import BitMap
+from iotdb.utils.NumpyTablet import NumpyTablet
+from iotdb.utils.Tablet import ColumnType, Tablet
+
+
+def _is_column_all_null_bitmap(bitmap: Optional[BitMap], row_number: int) ->
bool:
+ if bitmap is None or row_number <= 0:
+ return False
+ # BitMap is sized to maxRowNumber; only [0, row_number) are active rows.
+ return bitmap.is_range_all_marked(0, row_number)
+
+
+def _is_tablet_column_all_null(tablet: Tablet, column_index: int) -> bool:
+ values = tablet.get_values()
+ for row in range(tablet.get_row_number()):
+ if values[row][column_index] is not None:
+ return False
+ return tablet.get_row_number() > 0
+
+
+def filter_null_columns(
+ tablet: Union[Tablet, NumpyTablet],
+) -> Optional[Union[Tablet, NumpyTablet]]:
+ """
+ Drop entirely-null FIELD columns. TAG/ATTRIBUTE are always kept.
+ Does not mutate the input tablet.
+
+ Returns:
+ the same instance if nothing to drop;
+ a new tablet with remaining columns;
+ None if no columns remain. Table-model TAG / ATTRIBUTE columns are
kept even when
+ every FIELD column is null.
+ """
+ if tablet is None:
+ return None
+
+ column_types = tablet.get_column_categories()
+ column_number = len(tablet.get_measurements())
+ row_number = tablet.get_row_number()
+
+ kept_indices: List[int] = []
+
+ is_numpy = isinstance(tablet, NumpyTablet)
+ bitmaps = tablet.bitmaps if is_numpy else None
+
+ for i in range(column_number):
+ category = (
+ column_types[i]
+ if column_types is not None and i < len(column_types)
+ else ColumnType.FIELD
+ )
+ is_field = category == ColumnType.FIELD
+
+ if is_field:
+ if is_numpy:
+ bitmap = (
+ bitmaps[i] if bitmaps is not None and i < len(bitmaps)
else None
+ )
+ drop = _is_column_all_null_bitmap(bitmap, row_number)
+ else:
+ drop = _is_tablet_column_all_null(tablet, i)
+ else:
+ drop = False
+
+ if drop:
+ continue
+
+ kept_indices.append(i)
+
+ if len(kept_indices) == column_number:
+ return tablet
+ if not kept_indices:
+ return None
+
+ measurements = [tablet.get_measurements()[i] for i in kept_indices]
+ data_types = [tablet.get_data_types()[i] for i in kept_indices]
+ kept_column_types = [column_types[i] for i in kept_indices]
+
+ if is_numpy:
+ values = [tablet.get_values()[i] for i in kept_indices]
+ kept_bitmaps = None
+ if bitmaps is not None:
+ kept_bitmaps = [
+ bitmaps[i] if i < len(bitmaps) else None for i in kept_indices
+ ]
+ return NumpyTablet(
+ tablet.get_insert_target_name(),
+ measurements,
+ data_types,
+ values,
+ tablet.get_timestamps(),
+ bitmaps=kept_bitmaps,
+ column_types=kept_column_types,
+ )
+
+ # Tablet stores row-oriented values
+ src_values = tablet.get_values()
+ values = []
+ for row in range(row_number):
+ values.append([src_values[row][i] for i in kept_indices])
+ return Tablet(
+ tablet.get_insert_target_name(),
+ measurements,
+ data_types,
+ values,
+ list(tablet.get_timestamps()),
+ column_types=kept_column_types,
+ )
diff --git a/iotdb-client/client-py/iotdb/utils/Tablet.py
b/iotdb-client/client-py/iotdb/utils/Tablet.py
index 9b241723fe5..1239972486b 100644
--- a/iotdb-client/client-py/iotdb/utils/Tablet.py
+++ b/iotdb-client/client-py/iotdb/utils/Tablet.py
@@ -115,6 +115,12 @@ class Tablet(object):
def get_insert_target_name(self):
return self.__insert_target_name
+ def get_timestamps(self):
+ return self.__timestamps
+
+ def get_values(self):
+ return self.__values
+
def get_binary_timestamps(self):
format_str_list = [">"]
values_tobe_packed = []
diff --git a/iotdb-client/client-py/tests/unit/test_session_utils.py
b/iotdb-client/client-py/tests/unit/test_session_utils.py
new file mode 100644
index 00000000000..0af54b74320
--- /dev/null
+++ b/iotdb-client/client-py/tests/unit/test_session_utils.py
@@ -0,0 +1,160 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+#
+
+import numpy as np
+
+from iotdb.utils.BitMap import BitMap
+from iotdb.utils.IoTDBConstants import TSDataType
+from iotdb.utils.NumpyTablet import NumpyTablet
+from iotdb.utils.SessionUtils import filter_null_columns
+from iotdb.utils.Tablet import ColumnType, Tablet
+
+
+def test_filter_null_columns_tablet():
+ measurements = ["s1", "s2", "s3", "s4", "s5"]
+ data_types = [
+ TSDataType.INT32,
+ TSDataType.INT64,
+ TSDataType.FLOAT,
+ TSDataType.DOUBLE,
+ TSDataType.BOOLEAN,
+ ]
+ values = [[1, None, 1.5, None, None]]
+ timestamps = [1000]
+ tablet = Tablet("root.sg.d1", measurements, data_types, values, timestamps)
+
+ filtered = filter_null_columns(tablet)
+ assert filtered is not None
+ assert filtered is not tablet
+ assert filtered.get_measurements() == ["s1", "s3"]
+ assert filtered.get_values()[0] == [1, 1.5]
+
+ # nothing to drop
+ dense = Tablet(
+ "root.sg.d1",
+ ["s1"],
+ [TSDataType.INT32],
+ [[1]],
+ [1],
+ )
+ assert filter_null_columns(dense) is dense
+
+ # all null
+ all_null = Tablet(
+ "root.sg.d1",
+ measurements,
+ data_types,
+ [[None, None, None, None, None]],
+ [2000],
+ )
+ assert filter_null_columns(all_null) is None
+
+
+def test_filter_null_columns_table_model_keeps_tag():
+ tablet = Tablet(
+ "table1",
+ ["tag1", "s1", "s2"],
+ [TSDataType.STRING, TSDataType.INT32, TSDataType.INT32],
+ [["d1", None, None]],
+ [3000],
+ column_types=[ColumnType.TAG, ColumnType.FIELD, ColumnType.FIELD],
+ )
+ filtered = filter_null_columns(tablet)
+ assert filtered is not None
+ assert filtered is not tablet
+ assert filtered.get_measurements() == ["tag1"]
+ assert filtered.get_column_categories() == [ColumnType.TAG]
+
+
+def test_filter_null_columns_numpy_tablet():
+ measurements = ["s1", "s2", "s3"]
+ data_types = [TSDataType.INT32, TSDataType.INT64, TSDataType.FLOAT]
+ values = [
+ np.array([1], dtype=np.dtype(">i4")),
+ np.array([0], dtype=np.dtype(">i8")),
+ np.array([1.5], dtype=np.dtype(">f4")),
+ ]
+ timestamps = np.array([1000], dtype=np.dtype(">i8"))
+ bitmaps = [BitMap(1), BitMap(1), BitMap(1)]
+ bitmaps[1].mark(0)
+
+ np_tablet = NumpyTablet(
+ "root.sg.d1",
+ measurements,
+ data_types,
+ values,
+ timestamps,
+ bitmaps=bitmaps,
+ )
+ filtered = filter_null_columns(np_tablet)
+ assert filtered is not None
+ assert filtered is not np_tablet
+ assert filtered.get_measurements() == ["s1", "s3"]
+
+
+def test_filter_null_columns_numpy_tablet_active_rows():
+ measurements = ["s1", "s2"]
+ data_types = [TSDataType.INT32, TSDataType.INT64]
+ values = [
+ np.array([1], dtype=np.dtype(">i4")),
+ np.array([0], dtype=np.dtype(">i8")),
+ ]
+ timestamps = np.array([1000], dtype=np.dtype(">i8"))
+ bitmaps = [BitMap(10), BitMap(10)]
+ bitmaps[1].mark(0)
+
+ np_tablet = NumpyTablet(
+ "root.sg.d1",
+ measurements,
+ data_types,
+ values,
+ timestamps,
+ bitmaps=bitmaps,
+ )
+ filtered = filter_null_columns(np_tablet)
+ assert filtered is not None
+ assert filtered is not np_tablet
+ assert filtered.get_measurements() == ["s1"]
+
+
+def test_bitmap_is_range_all_marked():
+ bit_map = BitMap(20)
+ for i in range(20):
+ if i % 4 != 2:
+ bit_map.mark(i)
+ for start in range(21):
+ for length in range(21 - start):
+ all_marked = all(bit_map.is_marked(i) for i in range(start, start
+ length))
+ assert bit_map.is_range_all_marked(start, length) is all_marked
+ assert bit_map.is_range_all_marked(0, 0) is True
+ assert bit_map.is_range_all_marked(0, 21) is False
+ assert bit_map.is_range_all_marked(21, 0) is False
+ assert bit_map.is_range_all_marked(-1, 1) is False
+ assert bit_map.is_range_all_marked(0, -1) is False
+
+ empty = BitMap(0)
+ assert empty.is_all_marked() is True
+ full = BitMap(8)
+ for i in range(8):
+ full.mark(i)
+ assert full.is_all_marked() is True
+ partial = BitMap(8)
+ for i in range(7):
+ partial.mark(i)
+ assert partial.is_all_marked() is False
+ assert partial.is_range_all_marked(0, 7) is True
diff --git
a/iotdb-client/session/src/main/java/org/apache/iotdb/session/Session.java
b/iotdb-client/session/src/main/java/org/apache/iotdb/session/Session.java
index 36581b9690c..39bb9c06cba 100644
--- a/iotdb-client/session/src/main/java/org/apache/iotdb/session/Session.java
+++ b/iotdb-client/session/src/main/java/org/apache/iotdb/session/Session.java
@@ -2806,6 +2806,14 @@ public class Session implements ISession {
public void insertTablet(Tablet tablet, boolean sorted)
throws IoTDBConnectionException, StatementExecutionException {
TSInsertTabletReq request = genTSInsertTabletReq(tablet, sorted, false);
+ if (request == null) {
+ logger.warn(
+ ALL_VALUES_ARE_NULL,
+ tablet.getDeviceId(),
+ tablet.getRowSize() > 0 ? tablet.getTimestamp(0) : null,
+ tablet.getSchemas());
+ return;
+ }
insertTabletInternal(tablet, request);
}
@@ -2849,8 +2857,15 @@ public class Session implements ISession {
insertRelationalTabletWithLeaderCache(tablet);
} else {
TSInsertTabletReq request = genTSInsertTabletReq(tablet, false, false);
+ if (request == null) {
+ logger.warn(
+ ALL_VALUES_ARE_NULL,
+ tablet.getDeviceId(),
+ tablet.getRowSize() > 0 ? tablet.getTimestamp(0) : null,
+ tablet.getSchemas());
+ return;
+ }
request.setWriteToTable(true);
-
request.setColumnCategories(toEnumOrdinalsAsBytes(tablet.getColumnTypes()));
try {
getDefaultSessionConnection().insertTablet(request);
} catch (RedirectException ignored) {
@@ -2915,8 +2930,15 @@ public class Session implements ISession {
SessionConnection connection = entry.getKey();
Tablet tablet = entry.getValue();
TSInsertTabletReq request = genTSInsertTabletReq(tablet, false, false);
+ if (request == null) {
+ logger.warn(
+ ALL_VALUES_ARE_NULL,
+ tablet.getDeviceId(),
+ tablet.getRowSize() > 0 ? tablet.getTimestamp(0) : null,
+ tablet.getSchemas());
+ return;
+ }
request.setWriteToTable(true);
-
request.setColumnCategories(toEnumOrdinalsAsBytes(tablet.getColumnTypes()));
try {
connection.insertTablet(request);
} catch (RedirectException e) {
@@ -2956,9 +2978,15 @@ public class Session implements ISession {
return CompletableFuture.runAsync(
() -> {
TSInsertTabletReq request =
genTSInsertTabletReq(subTablet, false, false);
+ if (request == null) {
+ logger.warn(
+ ALL_VALUES_ARE_NULL,
+ subTablet.getDeviceId(),
+ subTablet.getRowSize() > 0 ?
subTablet.getTimestamp(0) : null,
+ subTablet.getSchemas());
+ return;
+ }
request.setWriteToTable(true);
- request.setColumnCategories(
- toEnumOrdinalsAsBytes(subTablet.getColumnTypes()));
InsertConsumer<TSInsertTabletReq> insertConsumer =
SessionConnection::insertTablet;
try {
@@ -3037,6 +3065,14 @@ public class Session implements ISession {
public void insertAlignedTablet(Tablet tablet, boolean sorted)
throws IoTDBConnectionException, StatementExecutionException {
TSInsertTabletReq request = genTSInsertTabletReq(tablet, sorted, true);
+ if (request == null) {
+ logger.warn(
+ ALL_VALUES_ARE_NULL,
+ tablet.getDeviceId(),
+ tablet.getRowSize() > 0 ? tablet.getTimestamp(0) : null,
+ tablet.getSchemas());
+ return;
+ }
try {
getSessionConnection(tablet.getDeviceId()).insertTablet(request);
} catch (RedirectException e) {
@@ -3065,9 +3101,14 @@ public class Session implements ISession {
sortTablet(tablet);
}
+ Tablet filtered = SessionUtils.filterNullColumns(tablet);
+ if (filtered == null) {
+ return null;
+ }
+
TSInsertTabletReq request = new TSInsertTabletReq();
- for (IMeasurementSchema measurementSchema : tablet.getSchemas()) {
+ for (IMeasurementSchema measurementSchema : filtered.getSchemas()) {
if (measurementSchema.getMeasurementName() == null) {
throw new
IllegalArgumentException(SessionMessages.MEASUREMENT_NON_NULL);
}
@@ -3075,22 +3116,25 @@ public class Session implements ISession {
request.addToTypes(measurementSchema.getType().ordinal());
}
- request.setPrefixPath(tablet.getDeviceId());
+ request.setPrefixPath(filtered.getDeviceId());
request.setIsAligned(isAligned);
+ if (filtered.getColumnTypes() != null) {
+
request.setColumnCategories(toEnumOrdinalsAsBytes(filtered.getColumnTypes()));
+ }
boolean trulyEnableRpcCompression =
- enableIoTDBRpcCompression && tablet.getRowSize() >=
tabletCompressionMinRowSize;
+ enableIoTDBRpcCompression && filtered.getRowSize() >=
tabletCompressionMinRowSize;
List<Byte> encodingTypes;
if (trulyEnableRpcCompression) {
- encodingTypes = new ArrayList<>(tablet.getSchemas().size() + 1);
+ encodingTypes = new ArrayList<>(filtered.getSchemas().size() + 1);
encodingTypes.add(
this.columnEncodersMap
.getOrDefault(
TSDataType.INT64,
TSEncoding.valueOf(TSFileDescriptor.getInstance().getConfig().getTimeEncoder()))
.serialize());
- for (IMeasurementSchema measurementSchema : tablet.getSchemas()) {
+ for (IMeasurementSchema measurementSchema : filtered.getSchemas()) {
if (measurementSchema.getMeasurementName() == null) {
throw new
IllegalArgumentException(SessionMessages.MEASUREMENT_NON_NULL);
}
@@ -3105,7 +3149,7 @@ public class Session implements ISession {
}
} else {
encodingTypes =
- Collections.nCopies(tablet.getSchemas().size() + 1,
TSEncoding.PLAIN.serialize());
+ Collections.nCopies(filtered.getSchemas().size() + 1,
TSEncoding.PLAIN.serialize());
}
TabletEncoder encoder =
@@ -3118,10 +3162,10 @@ public class Session implements ISession {
request.setCompressType(compressionType.serialize());
request.setEncodingTypes(encodingTypes);
}
- request.setTimestamps(encoder.encodeTime(tablet));
- request.setValues(encoder.encodeValues(tablet));
+ request.setTimestamps(encoder.encodeTime(filtered));
+ request.setValues(encoder.encodeValues(filtered));
- request.setSize(tablet.getRowSize());
+ request.setSize(filtered.getRowSize());
return request;
}
@@ -3154,6 +3198,9 @@ public class Session implements ISession {
} else {
TSInsertTabletsReq request =
genTSInsertTabletsReq(new ArrayList<>(tablets.values()), sorted,
false);
+ if (request == null) {
+ return;
+ }
try {
getDefaultSessionConnection().insertTablets(request);
} catch (RedirectException ignored) {
@@ -3190,6 +3237,9 @@ public class Session implements ISession {
} else {
TSInsertTabletsReq request =
genTSInsertTabletsReq(new ArrayList<>(tablets.values()), sorted,
true);
+ if (request == null) {
+ return;
+ }
try {
getDefaultSessionConnection().insertTablets(request);
} catch (RedirectException ignored) {
@@ -3208,6 +3258,11 @@ public class Session implements ISession {
updateTSInsertTabletsReq(request, entry.getValue(), sorted, isAligned);
}
+ tabletGroup.entrySet().removeIf(e -> e.getValue().getPrefixPathsSize() ==
0);
+ if (tabletGroup.isEmpty()) {
+ return;
+ }
+
if (tabletGroup.size() == 1) {
insertOnce(tabletGroup, SessionConnection::insertTablets);
} else {
@@ -3224,6 +3279,9 @@ public class Session implements ISession {
for (Tablet tablet : tablets) {
updateTSInsertTabletsReq(request, tablet, sorted, isAligned);
}
+ if (request.getPrefixPathsSize() == 0) {
+ return null;
+ }
return request;
}
@@ -3232,11 +3290,20 @@ public class Session implements ISession {
if (!checkSorted(tablet)) {
sortTablet(tablet);
}
- request.addToPrefixPaths(tablet.getDeviceId());
+ Tablet filtered = SessionUtils.filterNullColumns(tablet);
+ if (filtered == null) {
+ logger.warn(
+ ALL_VALUES_ARE_NULL,
+ tablet.getDeviceId(),
+ tablet.getRowSize() > 0 ? tablet.getTimestamp(0) : null,
+ tablet.getSchemas());
+ return;
+ }
+ request.addToPrefixPaths(filtered.getDeviceId());
List<String> measurements = new ArrayList<>();
List<Integer> dataTypes = new ArrayList<>();
request.setIsAligned(isAligned);
- for (IMeasurementSchema measurementSchema : tablet.getSchemas()) {
+ for (IMeasurementSchema measurementSchema : filtered.getSchemas()) {
if (measurementSchema.getMeasurementName() == null) {
throw new
IllegalArgumentException(SessionMessages.MEASUREMENT_NON_NULL);
}
@@ -3245,9 +3312,9 @@ public class Session implements ISession {
}
request.addToMeasurementsList(measurements);
request.addToTypesList(dataTypes);
- request.addToTimestampsList(SessionUtils.getTimeBuffer(tablet));
- request.addToValuesList(SessionUtils.getValueBuffer(tablet));
- request.addToSizeList(tablet.getRowSize());
+ request.addToTimestampsList(SessionUtils.getTimeBuffer(filtered));
+ request.addToValuesList(SessionUtils.getValueBuffer(filtered));
+ request.addToSizeList(filtered.getRowSize());
}
// sample some records and judge whether need to add too many null values to
convert to tablet.
@@ -3465,6 +3532,14 @@ public class Session implements ISession {
public void testInsertTablet(Tablet tablet, boolean sorted)
throws IoTDBConnectionException, StatementExecutionException {
TSInsertTabletReq request = genTSInsertTabletReq(tablet, sorted, false);
+ if (request == null) {
+ logger.warn(
+ ALL_VALUES_ARE_NULL,
+ tablet.getDeviceId(),
+ tablet.getRowSize() > 0 ? tablet.getTimestamp(0) : null,
+ tablet.getSchemas());
+ return;
+ }
getDefaultSessionConnection().testInsertTablet(request);
}
@@ -3487,6 +3562,9 @@ public class Session implements ISession {
throws IoTDBConnectionException, StatementExecutionException {
TSInsertTabletsReq request =
genTSInsertTabletsReq(new ArrayList<>(tablets.values()), sorted,
false);
+ if (request == null) {
+ return;
+ }
getDefaultSessionConnection().testInsertTablets(request);
}
diff --git
a/iotdb-client/session/src/main/java/org/apache/iotdb/session/util/SessionUtils.java
b/iotdb-client/session/src/main/java/org/apache/iotdb/session/util/SessionUtils.java
index 26a2e46e59d..b3b29f0be93 100644
---
a/iotdb-client/session/src/main/java/org/apache/iotdb/session/util/SessionUtils.java
+++
b/iotdb-client/session/src/main/java/org/apache/iotdb/session/util/SessionUtils.java
@@ -26,6 +26,7 @@ import org.apache.iotdb.session.i18n.SessionMessages;
import org.apache.tsfile.common.conf.TSFileConfig;
import org.apache.tsfile.encoding.encoder.Encoder;
+import org.apache.tsfile.enums.ColumnCategory;
import org.apache.tsfile.enums.TSDataType;
import org.apache.tsfile.file.metadata.IDeviceID;
import org.apache.tsfile.utils.Binary;
@@ -467,6 +468,91 @@ public class SessionUtils {
}
}
+ /**
+ * Remove FIELD columns that are entirely null within {@code [0, rowSize)}
according to BitMap.
+ * TAG / ATTRIBUTE columns are always kept. Does not mutate the input tablet.
+ *
+ * @param tablet source tablet
+ * @return the same instance if nothing to drop; a new tablet with remaining
columns; or {@code
+ * null} if no columns remain (e.g. tree-model tablet whose FIELD
columns are all null). For
+ * table-model tablets, TAG / ATTRIBUTE columns are kept even when every
FIELD column is null.
+ */
+ public static Tablet filterNullColumns(Tablet tablet) {
+ if (tablet == null) {
+ return null;
+ }
+ BitMap[] bitMaps = tablet.getBitMaps();
+ if (bitMaps == null) {
+ return tablet;
+ }
+
+ List<IMeasurementSchema> schemas = tablet.getSchemas();
+ List<ColumnCategory> columnCategories = tablet.getColumnTypes();
+ Object[] values = tablet.getValues();
+ int columnCount = schemas.size();
+ int rowSize = tablet.getRowSize();
+
+ List<IMeasurementSchema> keptSchemas = new ArrayList<>(columnCount);
+ List<ColumnCategory> keptCategories =
+ columnCategories != null ? new ArrayList<>(columnCount) : null;
+ List<Object> keptValues = new ArrayList<>(columnCount);
+ List<BitMap> keptBitMaps = new ArrayList<>(columnCount);
+
+ for (int i = 0; i < columnCount; i++) {
+ ColumnCategory category =
+ columnCategories != null && i < columnCategories.size()
+ ? columnCategories.get(i)
+ : ColumnCategory.FIELD;
+ boolean isField = category == ColumnCategory.FIELD;
+
+ boolean drop =
+ isField
+ && schemas.get(i).getMeasurementName() != null
+ && i < bitMaps.length
+ && isColumnAllNull(bitMaps[i], rowSize);
+ if (drop) {
+ continue;
+ }
+
+ keptSchemas.add(schemas.get(i));
+ if (keptCategories != null) {
+ keptCategories.add(category);
+ }
+ keptValues.add(values[i]);
+ keptBitMaps.add(i < bitMaps.length ? bitMaps[i] : null);
+ }
+
+ if (keptSchemas.size() == columnCount) {
+ return tablet;
+ }
+ if (keptSchemas.isEmpty()) {
+ return null;
+ }
+
+ Object[] newValues = keptValues.toArray();
+ BitMap[] newBitMaps = keptBitMaps.toArray(new BitMap[0]);
+ if (keptCategories != null) {
+ return new Tablet(
+ tablet.getDeviceId(),
+ keptSchemas,
+ keptCategories,
+ tablet.getTimestamps(),
+ newValues,
+ newBitMaps,
+ rowSize);
+ }
+ return new Tablet(
+ tablet.getDeviceId(), keptSchemas, tablet.getTimestamps(), newValues,
newBitMaps, rowSize);
+ }
+
+ private static boolean isColumnAllNull(BitMap bitMap, int rowSize) {
+ if (bitMap == null || rowSize <= 0) {
+ return false;
+ }
+ // BitMap is sized to maxRowNumber; only [0, rowSize) are active rows.
+ return bitMap.isRangeAllMarked(0, rowSize);
+ }
+
/* Used for table model insert only. */
public static boolean isTabletContainsSingleDevice(Tablet tablet) {
if (tablet.getRowSize() == 1) {
diff --git
a/iotdb-client/session/src/test/java/org/apache/iotdb/session/util/SessionUtilsTest.java
b/iotdb-client/session/src/test/java/org/apache/iotdb/session/util/SessionUtilsTest.java
index ffdc5835dfc..b8524d4b311 100644
---
a/iotdb-client/session/src/test/java/org/apache/iotdb/session/util/SessionUtilsTest.java
+++
b/iotdb-client/session/src/test/java/org/apache/iotdb/session/util/SessionUtilsTest.java
@@ -22,6 +22,7 @@ package org.apache.iotdb.session.util;
import org.apache.iotdb.common.rpc.thrift.TEndPoint;
import org.apache.iotdb.rpc.IoTDBConnectionException;
+import org.apache.tsfile.enums.ColumnCategory;
import org.apache.tsfile.enums.TSDataType;
import org.apache.tsfile.file.metadata.enums.CompressionType;
import org.apache.tsfile.file.metadata.enums.TSEncoding;
@@ -224,6 +225,88 @@ public class SessionUtilsTest {
() -> SessionUtils.getValueBuffer(typeList, valueList, measurements));
}
+ @Test
+ public void testFilterNullColumns() {
+ List<IMeasurementSchema> schemas = new ArrayList<>();
+ schemas.add(new MeasurementSchema("s1", TSDataType.INT32));
+ schemas.add(new MeasurementSchema("s2", TSDataType.INT64));
+ schemas.add(new MeasurementSchema("s3", TSDataType.FLOAT));
+ schemas.add(new MeasurementSchema("s4", TSDataType.DOUBLE));
+ schemas.add(new MeasurementSchema("s5", TSDataType.BOOLEAN));
+
+ Tablet tablet = new Tablet("root.sg.d1", schemas, 1);
+ tablet.addTimestamp(0, 1000L);
+ tablet.addValue("s1", 0, 1);
+ tablet.addValue("s2", 0, null);
+ tablet.addValue("s3", 0, 1.5f);
+ tablet.addValue("s4", 0, null);
+ tablet.addValue("s5", 0, null);
+
+ Tablet filtered = SessionUtils.filterNullColumns(tablet);
+ Assert.assertNotNull(filtered);
+ Assert.assertNotSame(tablet, filtered);
+ Assert.assertEquals(2, filtered.getSchemas().size());
+ Assert.assertEquals("s1",
filtered.getSchemas().get(0).getMeasurementName());
+ Assert.assertEquals("s3",
filtered.getSchemas().get(1).getMeasurementName());
+ Assert.assertEquals(1, ((int[]) filtered.getValues()[0])[0]);
+ Assert.assertEquals(1.5f, ((float[]) filtered.getValues()[1])[0], 0.0001f);
+
+ // no BitMap -> same instance
+ long[] timestamps = new long[] {1L};
+ Object[] values = new Object[] {new int[] {1}};
+ List<IMeasurementSchema> singleSchema =
+ Collections.singletonList(new MeasurementSchema("s1",
TSDataType.INT32));
+ Tablet noBitMapTablet = new Tablet("root.sg.d1", singleSchema, timestamps,
values, null, 1);
+ Assert.assertSame(noBitMapTablet,
SessionUtils.filterNullColumns(noBitMapTablet));
+
+ // all columns null -> null
+ Tablet allNull = new Tablet("root.sg.d1", schemas, 1);
+ allNull.addTimestamp(0, 2000L);
+ allNull.addValue("s1", 0, null);
+ allNull.addValue("s2", 0, null);
+ allNull.addValue("s3", 0, null);
+ allNull.addValue("s4", 0, null);
+ allNull.addValue("s5", 0, null);
+ Assert.assertNull(SessionUtils.filterNullColumns(allNull));
+
+ // table model: keep TAG when all FIELD columns are null
+ List<String> tableMeasurements = Arrays.asList("tag1", "s1", "s2");
+ List<TSDataType> tableDataTypes =
+ Arrays.asList(TSDataType.STRING, TSDataType.INT32, TSDataType.INT32);
+ List<ColumnCategory> columnCategories = new ArrayList<>();
+ columnCategories.add(ColumnCategory.TAG);
+ columnCategories.add(ColumnCategory.FIELD);
+ columnCategories.add(ColumnCategory.FIELD);
+ Tablet tableModelTablet =
+ new Tablet("table1", tableMeasurements, tableDataTypes,
columnCategories, 1);
+ tableModelTablet.addTimestamp(0, 3000L);
+ tableModelTablet.addValue("tag1", 0, "d1");
+ tableModelTablet.addValue("s1", 0, null);
+ tableModelTablet.addValue("s2", 0, null);
+ Tablet tableModelFiltered =
SessionUtils.filterNullColumns(tableModelTablet);
+ Assert.assertNotNull(tableModelFiltered);
+ Assert.assertNotSame(tableModelTablet, tableModelFiltered);
+ Assert.assertEquals(1, tableModelFiltered.getSchemas().size());
+ Assert.assertEquals("tag1",
tableModelFiltered.getSchemas().get(0).getMeasurementName());
+ Assert.assertEquals(ColumnCategory.TAG,
tableModelFiltered.getColumnTypes().get(0));
+ }
+
+ @Test
+ public void testFilterNullColumnsActiveRowsWhenBitmapSizedToMaxRowNumber() {
+ List<IMeasurementSchema> twoSchemas = new ArrayList<>();
+ twoSchemas.add(new MeasurementSchema("s1", TSDataType.INT32));
+ twoSchemas.add(new MeasurementSchema("s2", TSDataType.INT64));
+ Tablet partialRowTablet = new Tablet("root.sg.d1", twoSchemas, 10);
+ partialRowTablet.addTimestamp(0, 4000L);
+ partialRowTablet.addValue("s1", 0, 1);
+ partialRowTablet.addValue("s2", 0, null);
+ Tablet partialRowFiltered =
SessionUtils.filterNullColumns(partialRowTablet);
+ Assert.assertNotNull(partialRowFiltered);
+ Assert.assertNotSame(partialRowTablet, partialRowFiltered);
+ Assert.assertEquals(1, partialRowFiltered.getSchemas().size());
+ Assert.assertEquals("s1",
partialRowFiltered.getSchemas().get(0).getMeasurementName());
+ }
+
@Test
public void testParseSeedNodeUrls() {
List<String> nodeUrls = Collections.singletonList("127.0.0.1:1234");
