Fix XPRESS compression and enable in CI (#13649)

Summary:
Somehow this was previously not being tested in our Windows CI jobs so was accidentally broken in #13540  This fix will need to be backported to 10.3.

Pull Request resolved: #13649

Test Plan: CI

Reviewed By: hx235

Differential Revision: D75655418

Pulled By: pdillinger

fbshipit-source-id: a56bb213270904a1b7a13b905c2cc1919116df1c

Author: pdillinger

.github/actions/windows-build-steps/action.yml

@@ -38,7 +38,7 @@ runs:
       $env:Path = $env:JAVA_HOME + ";" + $env:Path
       mkdir build
       cd build
-      & cmake -G "$Env:CMAKE_GENERATOR" -DCMAKE_BUILD_TYPE=Debug -DOPTDBG=1 -DPORTABLE="$Env:CMAKE_PORTABLE" -DSNAPPY=1 -DJNI=1 ..
+      & cmake -G "$Env:CMAKE_GENERATOR" -DCMAKE_BUILD_TYPE=Debug -DOPTDBG=1 -DPORTABLE="$Env:CMAKE_PORTABLE" -DSNAPPY=1 -DXPRESS=1 -DJNI=1 ..
       if(!$?) { Exit $LASTEXITCODE }
       cd ..
       echo "Building with VS version: $Env:CMAKE_GENERATOR"

port/win/xpress_win.cc

@@ -202,11 +202,69 @@ char* Decompress(const char* input_data, size_t input_length,
   return outputBuffer.release();
 }
 
+int64_t GetDecompressedSize(const char* input_data, size_t input_length) {
+  assert(input_data != nullptr);
+
+  if (input_length == 0) {
+    return 0;
+  }
+
+  COMPRESS_ALLOCATION_ROUTINES* allocRoutinesPtr = nullptr;
+
+  DECOMPRESSOR_HANDLE decompressor = NULL;
+
+  BOOL success =
+      CreateDecompressor(COMPRESS_ALGORITHM_XPRESS,  //  Compression Algorithm
+                         allocRoutinesPtr,  //  Optional allocation routine
+                         &decompressor);    //  Handle
+
+  if (!success) {
+#ifdef _DEBUG
+    std::cerr << "XPRESS: Failed to create Decompressor LastError "
+              << GetLastError() << std::endl;
+#endif
+    return -1;
+  }
+
+  std::unique_ptr<void, decltype(CloseDecompressorFun)> decompressorGuard(
+      decompressor, CloseDecompressorFun);
+
+  SIZE_T decompressedBufferSize = 0;
+
+  success = ::Decompress(decompressor,                   //  Compressor Handle
+                         const_cast<char*>(input_data),  //  Compressed data
+                         input_length,              //  Compressed data size
+                         NULL,                      //  Buffer set to NULL
+                         0,                         //  Buffer size set to 0
+                         &decompressedBufferSize);  //  Decompressed Data size
+
+  assert(!success);
+  auto lastError = GetLastError();
+
+  if (lastError != ERROR_INSUFFICIENT_BUFFER) {
+#ifdef _DEBUG
+    std::cerr
+        << "XPRESS: Failed to estimate decompressed buffer size LastError "
+        << lastError << std::endl;
+#endif
+    return -1;
+  }
+
+  assert(decompressedBufferSize > 0);
+  return static_cast<int64_t>(decompressedBufferSize);
+}
+
 int64_t DecompressToBuffer(const char* input, size_t input_length, char* output,
                            size_t output_length) {
   assert(input != nullptr);
   assert(output != nullptr);
 
+  if (input_length == 0) {
+    return 0;
+  }
+
+  COMPRESS_ALLOCATION_ROUTINES* allocRoutinesPtr = nullptr;
+
   DECOMPRESSOR_HANDLE decompressor = NULL;
 
   BOOL success =

port/win/xpress_win.h

@@ -22,6 +22,8 @@ bool Compress(const char* input, size_t length, std::string* output);
 char* Decompress(const char* input_data, size_t input_length,
                  size_t* uncompressed_size);
 
+int64_t GetDecompressedSize(const char* input, size_t input_length);
+
 int64_t DecompressToBuffer(const char* input, size_t input_length, char* output,
                            size_t output_length);
 

util/compression.cc

@@ -590,7 +590,7 @@ class BuiltinDecompressorV2 : public Decompressor {
   Status ExtractUncompressedSize(Args& args) override {
     assert(args.compression_type != kNoCompression);
     if (args.compression_type == kSnappyCompression) {
-      // Exception to encoding of uncompressed size
+      // 1st exception to encoding of uncompressed size
 #ifdef SNAPPY
       size_t uncompressed_length = 0;
       if (!snappy::GetUncompressedLength(args.compressed_data.data(),
@@ -603,6 +603,20 @@ class BuiltinDecompressorV2 : public Decompressor {
 #else
       return Status::NotSupported("Snappy not supported in this build");
 #endif
+    } else if (args.compression_type == kXpressCompression) {
+      // 2nd exception to encoding of uncompressed size
+#ifdef XPRESS
+      int64_t result = port::xpress::GetDecompressedSize(
+          args.compressed_data.data(), args.compressed_data.size());
+      if (result < 0) {
+        return Status::Corruption("Error reading XPRESS compressed length");
+      }
+      args.uncompressed_size = static_cast<size_t>(result);
+      return Status::OK();
+#else
+      return Status::NotSupported("XPRESS not supported in this build");
+#endif
+
     } else {
       // Extract encoded uncompressed size
       return Decompressor::ExtractUncompressedSize(args);

util/compression.h

@@ -1260,7 +1260,8 @@ inline std::string CompressionOptionsToString(
 // block. Also, decompressed sizes for LZ4 are encoded in platform-dependent
 // way.
 // 2 -- Zlib, BZip2 and LZ4 encode decompressed size as Varint32 just before the
-// start of compressed block. Snappy format is the same as version 1.
+// start of compressed block. Snappy and XPRESS instead extract the decompressed
+// size from the compressed block itself, same as version 1.
 
 inline bool Snappy_Compress(const CompressionInfo& /*info*/, const char* input,
                             size_t length, ::std::string* output) {