further refactoring

Author: evanevanevanevannnn

ydb/library/yql/providers/s3/compressors/brotli.cpp

@@ -76,7 +76,7 @@ bool TReadBuffer::nextImpl() {
                 ythrow TCodeLineException(TIssuesIds::KIKIMR_BAD_REQUEST) << "Brotli decoder failed to decompress buffer: "
                                     << BrotliDecoderErrorString(BrotliDecoderGetErrorCode(DecoderState_));
             case BROTLI_DECODER_RESULT_NEEDS_MORE_OUTPUT:
-                YQL_ENSURE(availableOut != OutBuffer.size(), "Buffer passed to read in Brotli decoder is too small");
+                YQL_ENSURE_CODELINE(availableOut != OutBuffer.size(),TIssuesIds::KIKIMR_BAD_REQUEST, "Buffer passed to read in Brotli decoder is too small");
                 break;
             default:
                 break;

ydb/library/yql/providers/s3/compressors/gz.cpp

@@ -116,7 +116,7 @@ class TCompressor : public TOutputQueue<> {
             Z_.avail_out = OutputBufferSize;
 
             const auto code = deflate(&Z_, done ? Z_FINISH : Z_BLOCK);
-            YQL_ENSURE((done ? Z_STREAM_END : Z_OK) == code, "code: " << code << ", error: " << GetErrMsg(Z_));
+            YQL_ENSURE_CODELINE((done ? Z_STREAM_END : Z_OK) == code, TIssuesIds::KIKIMR_BAD_REQUEST, "code: " << code << ", error: " << GetErrMsg(Z_));
 
             if (const auto size = OutputBufferSize - Z_.avail_out)
                 TOutputQueue::Push(TString(OutputBuffer.get(), size));

ydb/library/yql/providers/s3/compressors/lz4io.cpp

@@ -7,6 +7,8 @@
 #include <contrib/libs/lz4/lz4hc.h>
 
 #include <ydb/library/yql/utils/yql_panic.h>
+#include <ydb/library/yql/utils/exceptions.h>
+#include <ydb/library/yql/core/issue/protos/issue_id.pb.h>
 #include "output_queue_impl.h"
 
 namespace NYql {
@@ -53,7 +55,7 @@ EStreamType CheckMagic(const void* data) {
 
 EStreamType CheckMagic(NDB::ReadBuffer& input) {
     char data[4u];
-    YQL_ENSURE(input.read(data, sizeof(data)) == sizeof(data), "Buffer too small.");
+    YQL_ENSURE_CODELINE(input.read(data, sizeof(data)) == sizeof(data), TIssuesIds::KIKIMR_BAD_REQUEST, "Buffer too small.");
     return CheckMagic(data);
 }
 
@@ -62,7 +64,7 @@ EStreamType CheckMagic(NDB::ReadBuffer& input) {
 TReadBuffer::TReadBuffer(NDB::ReadBuffer& source)
     : NDB::ReadBuffer(nullptr, 0ULL), Source(source), StreamType(CheckMagic(Source)), Pos(0ULL), Remaining(0ULL)
 {
-    YQL_ENSURE(StreamType != EStreamType::Unknown, "Wrong magic.");
+    YQL_ENSURE_CODELINE(StreamType != EStreamType::Unknown, TIssuesIds::KIKIMR_BAD_REQUEST, "Wrong magic.");
     if (StreamType == EStreamType::Frame) {
         const auto errorCode = LZ4F_createDecompressionContext(&Ctx, LZ4F_VERSION);
         YQL_ENSURE(!LZ4F_isError(errorCode), "Can't create LZ4F context resource: " << LZ4F_getErrorName(errorCode));
@@ -75,7 +77,7 @@ TReadBuffer::TReadBuffer(NDB::ReadBuffer& source)
         WriteLE32(InBuffer.data(), Lz4ioMagicNumber);
 
         NextToLoad = LZ4F_decompress_usingDict(Ctx, OutBuffer.data(), &outSize, InBuffer.data(), &inSize, nullptr, 0ULL, nullptr);
-        YQL_ENSURE(!LZ4F_isError(NextToLoad), "Header error: " << LZ4F_getErrorName(NextToLoad));
+        YQL_ENSURE_CODELINE(!LZ4F_isError(NextToLoad), TIssuesIds::KIKIMR_BAD_REQUEST, "Header error: " << LZ4F_getErrorName(NextToLoad));
     }
 }
 
@@ -120,7 +122,7 @@ size_t TReadBuffer::DecompressFrame() {
         if (Pos >= Remaining) {
             for (auto toRead = NextToLoad; toRead > 0U;) {
                 const auto sizeCheck = Source.read(InBuffer.data() + NextToLoad - toRead, toRead);
-                YQL_ENSURE(sizeCheck > 0U && sizeCheck <= toRead, "Cannot access compressed block.");
+                YQL_ENSURE_CODELINE(sizeCheck > 0U && sizeCheck <= toRead, TIssuesIds::KIKIMR_BAD_REQUEST, "Cannot access compressed block.");
                 toRead -= sizeCheck;
             }
 
@@ -132,7 +134,7 @@ size_t TReadBuffer::DecompressFrame() {
         while (Pos < Remaining || (decodedBytes == OutBuffer.size())) {
             decodedBytes = OutBuffer.size();
             NextToLoad = LZ4F_decompress_usingDict(Ctx, OutBuffer.data(), &decodedBytes, InBuffer.data() + Pos, &Remaining, nullptr, 0ULL, nullptr);
-            YQL_ENSURE(!LZ4F_isError(NextToLoad), "Decompression error: " << LZ4F_getErrorName(NextToLoad));
+            YQL_ENSURE_CODELINE(!LZ4F_isError(NextToLoad), TIssuesIds::KIKIMR_BAD_REQUEST, "Decompression error: " << LZ4F_getErrorName(NextToLoad));
             Pos += Remaining;
 
             if (decodedBytes)
@@ -153,21 +155,21 @@ size_t TReadBuffer::DecompressLegacy() {
     unsigned int blockSize = 0U;
 
     if (const auto sizeCheck = Source.read(InBuffer.data(), 4U)) {
-        YQL_ENSURE(sizeCheck == 4U, "Cannot access block size.");
+        YQL_ENSURE_CODELINE(sizeCheck == 4U, TIssuesIds::KIKIMR_BAD_REQUEST, "Cannot access block size.");
         blockSize = ReadLE32(InBuffer.data());
-        YQL_ENSURE(blockSize <= LZ4_COMPRESSBOUND(LegacyBlockSize), "Block size out of bounds.");
+        YQL_ENSURE_CODELINE(blockSize <= LZ4_COMPRESSBOUND(LegacyBlockSize), TIssuesIds::KIKIMR_BAD_REQUEST, "Block size out of bounds.");
     } else
         return 0ULL;
 
     for (auto toRead = blockSize; toRead > 0U;) {
         const auto sizeCheck = Source.read(InBuffer.data() + blockSize - toRead, toRead);
-        YQL_ENSURE(sizeCheck > 0U && sizeCheck <= toRead, "Cannot access compressed block.");
+        YQL_ENSURE_CODELINE(sizeCheck > 0U && sizeCheck <= toRead, TIssuesIds::KIKIMR_BAD_REQUEST, "Cannot access compressed block.");
         toRead -= sizeCheck;
     }
 
     const auto decodeSize = LZ4_decompress_safe(InBuffer.data(), OutBuffer.data(), (int)blockSize, LegacyBlockSize);
 
-    YQL_ENSURE(decodeSize >= 0, "Corrupted input detected.");
+    YQL_ENSURE_CODELINE(decodeSize >= 0, TIssuesIds::KIKIMR_BAD_REQUEST, "Corrupted input detected.");
     return size_t(decodeSize);
 }
 

ydb/library/yql/providers/s3/compressors/zstd.cpp

@@ -2,6 +2,8 @@
 
 #include <util/generic/size_literals.h>
 #include <ydb/library/yql/utils/yql_panic.h>
+#include <ydb/library/yql/utils/exceptions.h>
+#include <ydb/library/yql/core/issue/protos/issue_id.pb.h>
 #include "output_queue_impl.h"
 
 namespace NYql {
@@ -34,13 +36,13 @@ bool TReadBuffer::nextImpl() {
             zIn.pos = Offset_ = 0;
             if (!zIn.size) {
                 // end of stream, need to check that there is no uncompleted blocks
-                YQL_ENSURE(!returnCode, "Incomplete block.");
+                YQL_ENSURE_CODELINE(!returnCode, TIssuesIds::KIKIMR_BAD_REQUEST, "Incomplete block.");
                 Finished_ = true;
                 break;
             }
         }
         returnCode = ::ZSTD_decompressStream(ZCtx_, &zOut, &zIn);
-        YQL_ENSURE(!::ZSTD_isError(returnCode), "Decompress failed: " << ::ZSTD_getErrorName(returnCode));
+        YQL_ENSURE_CODELINE(!::ZSTD_isError(returnCode), TIssuesIds::KIKIMR_BAD_REQUEST, "Decompress failed: " << ::ZSTD_getErrorName(returnCode));
         if (!returnCode) {
             // The frame is over, prepare to (maybe) start a new frame
             ::ZSTD_initDStream(ZCtx_);
@@ -65,7 +67,7 @@ class TCompressor : public TOutputQueue<> {
         : ZCtx_(::ZSTD_createCStream())
     {
         const auto ret = ::ZSTD_initCStream(ZCtx_, level);
-        YQL_ENSURE(!::ZSTD_isError(ret), "code: " << ret << ", error: " << ::ZSTD_getErrorName(ret));
+        YQL_ENSURE_CODELINE(!::ZSTD_isError(ret), TIssuesIds::KIKIMR_BAD_REQUEST, "code: " << ret << ", error: " << ::ZSTD_getErrorName(ret));
     }
 
     ~TCompressor() {

ydb/library/yql/utils/exceptions.h

@@ -24,4 +24,11 @@ struct TCodeLineException: public yexception {
 
 TCodeLineException operator+(const TSourceLocation& sl, TCodeLineException&& t);
 
-} // namespace NFq
+#define YQL_ENSURE_CODELINE(CONDITION, CODE, ...)     \
+    do {                                   \
+        if (Y_UNLIKELY(!(CONDITION))) {    \
+            ythrow TCodeLineException(CODE) << __VA_ARGS__; \
+        }                                  \
+    } while (0)
+
+} // namespace NYql