koturnの日記

普通の人です.ブログ上のコードはコピペ自由です.

Unityのシェーダーにおいて,定数の整数乗を行う場合の繰り返し二乗法はループ展開されるか?

TL;DR

繰り返し二乗法の関数を定義し,指数に定数を指定した場合はループ展開される. しかし,pow(x, 5.0) のように指数が比較的小さい整数定数を指定した場合でも繰り返し二乗法をループ展開したものと同一のコード生成がされるため,自前で繰り返し二乗法の関数を設ける必要はないかもしれない.

前置き

Unityのシェーダーの標準ライブラリのコードを眺めていて,以下のようなものを見つけた.

  • UnityStandardBRDF.cginc
inline half Pow4 (half x)
{
    return x*x*x*x;
}

inline float2 Pow4 (float2 x)
{
    return x*x*x*x;
}

inline half3 Pow4 (half3 x)
{
    return x*x*x*x;
}

inline half4 Pow4 (half4 x)
{
    return x*x*x*x;
}

// Pow5 uses the same amount of instructions as generic pow(), but has 2 advantages:
// 1) better instruction pipelining
// 2) no need to worry about NaNs
inline half Pow5 (half x)
{
    return x*x * x*x * x;
}

inline half2 Pow5 (half2 x)
{
    return x*x * x*x * x;
}

inline half3 Pow5 (half3 x)
{
    return x*x * x*x * x;
}

inline half4 Pow5 (half4 x)
{
    return x*x * x*x * x;
}

pow() 関数による累乗の計算は整数ではない指数による累乗にも対応している. しかし,実装としては log2()exp2() を利用したものとなるため,Pow5() のコメントにあるように,命令数は同じでも(乗算3回か log, mul, exp の3命令)計算負荷は単純な乗算の方が軽いはずである.

float3 pow(float3 x, float3 y)

    return exp2(log2(x) * y);
}

4乗と5乗で同じように処理を記述しているのはコード量が膨れ上がると思った. また,2乗や3乗のバリエーションも欲しい.

指数が非負整数の累乗は繰り返し二乗法で効率的に計算可能であることはよく知られている. 以下のように繰り返し二乗法の関数を定義し,累乗部分に定数を指定することで,良い感じにループ展開されるかどうかを調べた. (すなわち,Pow4()Pow5() の出力アセンブリと一致,または同等のものになるかどうか調べた)

inline float pown(float x, int n)
{
    float v = 1.0;
    UNITY_UNROLL
    for (; n > 0; n >>= 1) {
        v *= (n & 1) == 0 ? 1.0 : x;
        x *= x;
    }
    return v;
}

inline float2 pown(float2 x, int n)
{
    static const float2 ones = float2(1.0, 1.0);

    float2 v = ones;
    UNITY_UNROLL
    for (; n > 0; n >>= 1) {
        v *= (n & 1) == 0 ? ones : x;
        x *= x;
    }
    return v;
}

inline float3 pown(float3 x, int n)
{
    static const float3 ones = float3(1.0, 1.0, 1.0);

    float3 v = ones;
    UNITY_UNROLL
    for (; n > 0; n >>= 1) {
        v *= (n & 1) == 0 ? ones : x;
        x *= x;
    }
    return v;
}

inline float4 pown(float4 x, int n)
{
    static const float4 ones = float4(1.0, 1.0, 1.0, 1.0);

    float4 v = ones;
    UNITY_UNROLL
    for (; n > 0; n >>= 1) {
        v *= (n & 1) == 0 ? ones : x;
        x *= x;
    }
    return v;
}

繰り返し二乗法

繰り返し二乗法の再帰的な定義は下記の通り(簡単のため $n$ は非負整数とする).

\begin{equation} x^{n} = \begin{cases} x (x^{\frac{n - 1}{2}})^2 & \text{where} ~ n \equiv 1 \pmod 2 \\ (x^{\frac{n}{2}})^2 & \text{where} ~ n \equiv 0 \pmod 2 \end{cases} \end{equation}

乗算命令の回数は$O(\log n)$であるが,正確には下記の通り(ループ展開した場合).

\begin{equation} C(n) = \lfloor \log_2 n \rfloor + popcnt(n) - 1 \label{numberOfMulExpBySquaring} \end{equation}

$popcnt$ は立っているビット数を数える関数であり,あえて数式で記述するならば下記のような定義となる.

\begin{equation} popcnt(n) = \begin{cases} 0 & \text{where} ~ n = 0 \\ popcnt \left(\dfrac{n}{2} \right) & \text{where} ~ n > 0 ~ \text{and} ~ n \equiv 0 \pmod 2 \\ 1 + popcnt \left( \dfrac{n - 1}{2} \right) & \text{where} ~ n > 0 ~ \text{and} ~ n \equiv 1 \pmod 2 \end{cases} \end{equation}

確認用シェーダー

指数は5,ベクトルの次元は3で確認すれば十分だろう. multi_compile 用のキーワードを設け,それぞれ下記のコードになるようにした.

キーワード コード詳細
NAIVE pow() 組み込み関数を利用
NAIVE_MUL 自前の乗算5回を行う関数を利用
ITERATIVE_SQUARE 自前の繰り返し二乗法を行う関数を利用
  • IntPower.shader
Shader "koturn/IntPower"
{
    Properties
    {
        [KeywordEnum(NAIVE, NAIVE_MUL, ITERATIVE_SQUARE)]
        _ExpMethod("Exp method", Int) = 2  // Default: ITERATIVE_SQUARE
    }

    SubShader
    {
        Pass
        {
            CGPROGRAM
            #pragma vertex vert
            #pragma fragment frag
            #pragma target 3.0
            #pragma multi_compile_local_fragment _EXPMETHOD_NAIVE _EXPMETHOD_NAIVE_MUL _EXPMETHOD_ITERATIVE_SQUARE

            #include "UnityCG.cginc"

            struct appdata
            {
                float4 vertex : POSITION;
                float2 uv : TEXCOORD0;
                float4 color : COLOR;
            };

            struct v2f
            {
                float4 vertex : SV_POSITION;
                float2 uv : TEXCOORD0;
                float4 color : TEXCOORD1;
            };


            inline float3 pow5(float3 x)
            {
                return x * x * x * x * x;
            }

            inline float3 pown(float3 x, int n)
            {
                static const float3 ones = float3(1.0, 1.0, 1.0);

                float3 v = ones;
                UNITY_UNROLL
                for (; n > 0; n >>= 1) {
                    v *= (n & 1) == 0 ? ones : x;
                    x *= x;
                }
                return v;
            }

            v2f vert(appdata v)
            {
                v2f o;
                o.vertex = UnityObjectToClipPos(v.vertex);
                o.uv = v.uv;
                o.color = v.color;

                return o;
            }

            fixed4 frag(v2f i) : SV_Target
            {
#if defined(_EXPMETHOD_NAIVE)
                // return float4(pow(i.color.rgb, 5.0), 1.0);
                // return float4(exp2(log2(i.color.rgb) * 5.0)), 1.0);
                return float4(exp2(i.color.rgb * log2(5.0))), 1.0);
#elif defined(_EXPMETHOD_NAIVE_MUL)
                return float4(pow5(i.color.rgb), 1.0);
#else
                return float4(pown(i.color.rgb, 5), 1.0);
#endif
            }
            ENDCG
        }
    }
}

確認用シェーダーの生成コード

各キーワードに対し,それぞれ下記のコードが生成されていた. すなわち,どれも全く同じ繰り返し二乗法をループ展開したコードとなっていた. となると,繰り返し二乗法の関数を自前で定義する必要はないかもしれない?

  • _EXPMETHOD_NAIVE
//////////////////////////////////////////////////////
Global Keywords: <none>
Local Keywords: _EXPMETHOD_NAIVE
-- Vertex shader for "d3d11":
// No shader variant for this keyword set. The closest match will be used instead.

-- Hardware tier variant: Tier 1
-- Fragment shader for "d3d11":
// Stats: 3 math, 1 temp registers
Shader Disassembly:
//
// Generated by Microsoft (R) D3D Shader Disassembler
//
//
// Input signature:
//
// Name                 Index   Mask Register SysValue  Format   Used
// -------------------- ----- ------ -------- -------- ------- ------
// SV_POSITION              0   xyzw        0      POS   float
// TEXCOORD                 0   xy          1     NONE   float
// TEXCOORD                 1   xyzw        2     NONE   float   xyz
//
//
// Output signature:
//
// Name                 Index   Mask Register SysValue  Format   Used
// -------------------- ----- ------ -------- -------- ------- ------
// SV_Target                0   xyzw        0   TARGET   float   xyzw
//
      ps_4_0
      dcl_input_ps linear v2.xyz
      dcl_output o0.xyzw
      dcl_temps 1
   0: mul r0.xyz, v2.xyzx, v2.xyzx
   1: mul r0.xyz, r0.xyzx, r0.xyzx
   2: mul o0.xyz, r0.xyzx, v2.xyzx
   3: mov o0.w, l(1.000000)
   4: ret
// Approximately 0 instruction slots used
  • _EXPMETHOD_NAIVE_MUL
//////////////////////////////////////////////////////
Global Keywords: <none>
Local Keywords: _EXPMETHOD_NAIVE_MUL
-- Vertex shader for "d3d11":
// No shader variant for this keyword set. The closest match will be used instead.

-- Hardware tier variant: Tier 1
-- Fragment shader for "d3d11":
// Stats: 3 math, 1 temp registers
Shader Disassembly:
//
// Generated by Microsoft (R) D3D Shader Disassembler
//
//
// Input signature:
//
// Name                 Index   Mask Register SysValue  Format   Used
// -------------------- ----- ------ -------- -------- ------- ------
// SV_POSITION              0   xyzw        0      POS   float
// TEXCOORD                 0   xy          1     NONE   float
// TEXCOORD                 1   xyzw        2     NONE   float   xyz
//
//
// Output signature:
//
// Name                 Index   Mask Register SysValue  Format   Used
// -------------------- ----- ------ -------- -------- ------- ------
// SV_Target                0   xyzw        0   TARGET   float   xyzw
//
      ps_4_0
      dcl_input_ps linear v2.xyz
      dcl_output o0.xyzw
      dcl_temps 1
   0: mul r0.xyz, v2.xyzx, v2.xyzx
   1: mul r0.xyz, r0.xyzx, r0.xyzx
   2: mul o0.xyz, r0.xyzx, v2.xyzx
   3: mov o0.w, l(1.000000)
   4: ret
// Approximately 0 instruction slots used
  • _EXPMETHOD_ITERATIVE_SQUARE
//////////////////////////////////////////////////////
Global Keywords: <none>
Local Keywords: _EXPMETHOD_ITERATIVE_SQUARE
-- Vertex shader for "d3d11":
// No shader variant for this keyword set. The closest match will be used instead.

-- Hardware tier variant: Tier 1
-- Fragment shader for "d3d11":
// Stats: 3 math, 1 temp registers
Shader Disassembly:
//
// Generated by Microsoft (R) D3D Shader Disassembler
//
//
// Input signature:
//
// Name                 Index   Mask Register SysValue  Format   Used
// -------------------- ----- ------ -------- -------- ------- ------
// SV_POSITION              0   xyzw        0      POS   float
// TEXCOORD                 0   xy          1     NONE   float
// TEXCOORD                 1   xyzw        2     NONE   float   xyz
//
//
// Output signature:
//
// Name                 Index   Mask Register SysValue  Format   Used
// -------------------- ----- ------ -------- -------- ------- ------
// SV_Target                0   xyzw        0   TARGET   float   xyzw
//
      ps_4_0
      dcl_input_ps linear v2.xyz
      dcl_output o0.xyzw
      dcl_temps 1
   0: mul r0.xyz, v2.xyzx, v2.xyzx
   1: mul r0.xyz, r0.xyzx, r0.xyzx
   2: mul o0.xyz, r0.xyzx, v2.xyzx
   3: mov o0.w, l(1.000000)
   4: ret
// Approximately 0 instruction slots used

pow()関数はどこまでを繰り返し二乗法のコードとするか

繰り返し二乗法の乗算回数は式 \eqref{numberOfMulExpBySquaring} の通りであるため,かなり多い乗算回数よりも log, mul, exp の3命令を用いた方が好ましいと考えられる. シェーダーコンパイラが整数定数を指定した pow() 関数から log, mul, exp の3命令を生成する閾値を調べてみることにした.

結果は下記の表の通り. この結果から指数 $n$ ではなく,乗算命令数 $C(n)$ が閾値となっており,7回以下なら繰り返し二乗法,8回以上なら log, mul, exp の3命令を生成することがわかる.

$$n$$ $$C(n)$$ 複数の乗算命令?
1 0 movのみ
2 1
3 2
4 2
5 3
6 3
7 4
8 3
9 4
10 4
11 5
12 4
13 5
14 5
15 6
16 4
17 5
18 5
19 6
20 5
21 6
22 6
23 7
24 5
25 6
26 6
27 7
28 6
29 7
30 7
31 8
32 5

まとめ

繰り返し二乗法の関数を用意し,指数に定数を指定することで,単純な乗算の命令が並ぶことが確認できた. これにより,2乗,3乗,4乗,...の関数を個別に用意せずとも,繰り返し二乗法の関数だけ用意するだけでよいことになる.

しかし,Direct3D11の出力アセンブラを確認する限り,組み込み関数 pow() の指数に定数を指定することでも繰り返し二乗法のコード生成をすることがわかった. しかも,乗算命令数によっては log, mul, exp の3命令にするようだ. このことから,わざわざ自前で繰り返し二乗法の関数を用意する必要はないかもしれない.

参考文献

シェーダーにおけるゼロベクトルの正規化

シェーダーにおいて,組み込み関数である normalize() にゼロベクトルを渡した場合,NaN になる. 正規化ベクトルの定義からも当然である.

\begin{equation} normalize(\boldsymbol{v}) = \dfrac{\boldsymbol{v}}{\| \boldsymbol{v} \|} \end{equation}

シェーダーとしては normalize() は下記の実装と同等である. float3 を一例としているが,float2float4 等でも同様である.

float3 normalize(float3 v)
{
    return rsqrt(dot(v, v)) * v;
}

rsqrt() は逆数平方根の関数であり,高速に逆数平方根を計算する rsq 命令となるが,この命令は 0 に対し, Inf を返す. その結果, Inf0 の積となるため, normalize() の計算結果は NaN となる.

ゼロベクトルに対してはゼロベクトルを返すnormalizeを定義する

NaN は取り扱いづらいため,ゼロベクトルを正規化しようとした場合は特殊扱いでゼロベクトルを返すようにした方が都合が良いこともある. そこで,前述の normalize() の実装を少し変更する.

float3 normalizeEx(float3 v)
{
    const float vdotV = dot(v, v);
    return vdotV == 0.0 ? v : rsqrt(vdotV) * v;
}

ただ,基本的に浮動小数点数の等値比較は避けるべきであるのと,Unity C#Vector2, Vecotr3, Vector4normalized は長さが微小値(1.0e-5 以下)のベクトルに対してはゼロベクトルを返す実装となっているので,それに習い下記のようにした方がよいかもしれない.

float3 normalizeEx(float3 v)
{
    const float vdotV = dot(v, v);
    return vdotV <= 1.0e-10 ? float3(0.0, 0.0, 0.0) : rsqrt(vdotV) * v;
}

参考文献

シェーダーにおける浮動小数点剰余(mod, fmod)の実装

TL;DR

シェーダー言語によってその言語組み込みの fmod()mod()) の実装は異なるため注意する必要がある. 挙動の罠に引っかからないためには自前で実装するのが安全である.

GLSL

// Equivalent to mod() in GLSL.
float fmodglsl(float x, float y)
{
    return x - y * floor(x / y);
}

HLSL

// Equivalent to fmod() in HLSL.
float fmodhlsl(float x, float y)
{
    return x - y * trunc(x / y);
}

CG, Unityのshaderlab(CGPROGRAM, HLSLPROGRAM かに依存しない)

// Equivalent to fmod() in CG.
float fmodcg(float x, float y)
{
    const float c = frac(abs(x / y)) * abs(y);
    return x < 0 ? -c : c;
}

fmodについて

fmod() とは浮動小数点の剰余の計算を行う関数である. シェーダー特有のものではなく,C言語<math.h> にも存在しているような普遍的な計算である.

Unityのshaderlabでの確認方法

僕個人としてはshaderlabをいじることがほとんどである. そのため,shaderlabでの組み込み関数 fmod() の実装がどうなっているかを調べることにした.

実装を見るためには,組み込み関数の fmod() と自前で用意した剰余関数から生成される命令(Direct X11用)を比較するのが手っ取り早い. とりあえず,以下のコードで確認することにした.

FmodImpl.shader

Shader "koturn/FmodImpl"
{
    Properties
    {
        _MainTex ("Main texture", 2D) = "white" {}
        _Color ("Multiplicative color for _MainTex", Color) = (1.0, 1.0, 1.0, 1.0)

        _Value01 ("Value01", Float) = 23.0
        _Value02 ("Value02", Float) = 3.7

        [KeywordEnum(Operator, Native Func, CG, HLSL, GLSL)]
        _ModType ("Mod operation type", Float) = 2  // Default: Back

        [Enum(UnityEngine.Rendering.CullMode)]
        _Cull ("Cull", Float) = 2  // Default: Back

        [Enum(UnityEngine.Rendering.CompareFunction)]
        _ZTest ("ZTest", Float) = 4  // Default: LEqual

        [Enum(Off, 0, On, 1)]
        _ZWrite ("ZWrite", Float) = 0  // Default: Off

        [Enum(UnityEngine.Rendering.BlendMode)]
        _SrcFactor ("Src Factor", Float) = 5  // Default: SrcAlpha

        [Enum(UnityEngine.Rendering.BlendMode)]
        _DstFactor ("Dst Factor", Float) = 10  // Default: OneMinusSrcAlpha
    }

    SubShader
    {
        Tags
        {
            "RenderType" = "Transparent"
            "Queue" = "Transparent"
        }

        Cull [_Cull]
        ZWrite [_ZWrite]
        ZTest [_ZTest]
        Blend [_SrcFactor] [_DstFactor]

        Pass
        {
            CGPROGRAM
            #pragma target 3.0

            #pragma vertex vert
            #pragma fragment frag

            #pragma multi_compile_local_fragment _MODTYPE_OPERATOR _MODTYPE_NATIVE_FUNC _MODTYPE_CG _MODTYPE_HLSL _MODTYPE_GLSL

            #include "UnityCG.cginc"

            UNITY_DECLARE_TEX2D(_MainTex);
            uniform float4 _Color;
            uniform float _Value01;
            uniform float _Value02;


            float fmodcg(float x, float y)
            {
                const float c = frac(abs(x / y)) * abs(y);
                return x < 0 ? -c : c;
            }

            float fmodglsl(float x, float y)
            {
                return x - y * floor(x / y);
            }

            float fmodhlsl(float x, float y)
            {
                return x - y * trunc(x / y);
            }


            struct appdata
            {
                float4 vertex : POSITION;
                float2 uv : TEXCOORD0;
            };

            struct v2f
            {
                float4 vertex : SV_POSITION;
                float2 uv : TEXCOORD0;
            };


            v2f vert(appdata v)
            {
                v2f o;
                o.vertex = UnityObjectToClipPos(v.vertex);
                o.uv = v.uv;
                return o;
            }

            fixed4 frag(v2f i) : SV_Target
            {
                float4 col = UNITY_SAMPLE_TEX2D(_MainTex, i.uv) * _Color;
#if defined(_MODTYPE_NATIVE_FUNC)
                col.a = fmod(_Value01, _Value02);
#elif defined(_MODTYPE_OPERATOR)
                col.a = _Value01 % _Value02;
#elif defined(_MODTYPE_CG)
                col.a = fmodcg(_Value01, _Value02);
#elif defined(_MODTYPE_HLSL)
                col.a = fmodhlsl(_Value01, _Value02);
#elif defined(_MODTYPE_GLSL)
                col.a = fmodglsl(_Value01, _Value02);
#endif
                return col;
            }
            ENDCG
        }
    }
}

生成コード

部分的に抜粋する.

組み込み関数のfmod

今回の検証対象である. 生成された命令からNVidiaのCgの言語仕様に書かれているのと同様の実装になっていることがわかる.

   0: div r0.x, cb0[3].x, cb0[3].y
   1: ge r0.y, r0.x, -r0.x
   2: frc r0.x, |r0.x|
   3: movc r0.x, r0.y, r0.x, -r0.x
   4: mul o0.w, r0.x, cb0[3].y

また,CGPROGRAMHLSLPROGRAM かに依らず同じコードが生成された. 挙動をCgの実装に統一しているのだろう.

演算子

演算子 % を用いた場合でもコンパイルは通るのだが,組み込み関数の fmod() を用いた場合とは異なるコードが生成されていた.

   0: mul r0.x, cb0[3].y, cb0[3].x
   1: ge r0.x, r0.x, -r0.x
   2: movc r0.x, r0.x, cb0[3].y, -cb0[3].y
   3: div r0.y, l(1.000000, 1.000000, 1.000000, 1.000000), r0.x
   4: mul r0.y, r0.y, cb0[3].x
   5: frc r0.y, r0.y
   6: mul o0.w, r0.y, r0.x

これをシェーダーコードに翻訳すると下記のようになる.

float fmodop(float x, float y)
{
    const float xy = x * y;
    const float a = (xy >= -xy) ? y : -y;
    const float b = frac((1.0 / a) * x);
    return a * b;
}

計算結果は浮動小数点計算における誤差を無視すれば,組み込み関数の fmod() と一致する. 生成されているコード量としてあまり効率的でないように思われるが実際はどうなのだろうか?

例えば,

   3: div r0.y, l(1.000000, 1.000000, 1.000000, 1.000000), r0.x
   4: mul r0.y, r0.y, cb0[3].x

   div r0.y, cb0[3].x, r0.x

でいいのでは?などと思ったりする.

fmodglsl

float fmodglsl(float x, float y)
{
    return x - y * floor(x / y);
}
   0: div r0.x, cb0[3].x, cb0[3].y
   1: round_ni r0.x, r0.x
   2: mad o0.w, -cb0[3].y, r0.x, cb0[3].x

floor()round_ni 命令に対応しており,命令名はおそらくRound to Negative Infinityのことで,負の無限大への丸めを意味するのだろう. y が正のときは正の値を,y が負のときは負の値を返すようになっている. x / y が負数のときの挙動が fmodhlsl() と異なり,正負に関係なく周期的な挙動をするため,素直に扱いやすいと思う.

fmodhlsl

float fmodhlsl(float x, float y)
{
    return x - y * trunc(x / y);
}
   0: div r0.x, cb0[3].x, cb0[3].y
   1: round_z r0.x, r0.x
   2: mad o0.w, -cb0[3].y, r0.x, cb0[3].x

trunc()round_z 命令に対応しており,命令名はおそらくRound to Zeroのことで,0への丸めを意味するのだろう. x / y が負数のときの挙動が fmodglsl() と異なり,正負で周期的な挙動が切り替わる形である.

y の符号が反転したとしても同じ値を返す.

fmodcg

float fmodcg(float x, float y)
{
    const float c = frac(abs(x / y)) * abs(y);
    return x < 0 ? -c : c;
}

生成される命令としては組み込み関数の fmod() と一致している. 前述の2つよりも命令数が少し多い.

   0: div r0.x, cb0[3].x, cb0[3].y
   1: frc r0.x, |r0.x|
   2: mul r0.x, r0.x, |cb0[3].y|
   3: lt r0.y, cb0[3].x, l(0.000000)
   4: movc o0.w, r0.y, -r0.x, r0.x

誤差を考慮しなければ算出される値は fmodhlsl() と同じである.

ただし,前述の2つと比較すると,正または負に誤差が発生するので,この関数の返却値に対して積・商を行った結果をそのまま floor()ceil() に通すと想定外の値となることがある. (fmodglslfmodhlsl を用いた場合と異なる結果になる)

あまりこの実装を用いるメリットはないように思われるが....

参考文献

PostgreSQLで外部テーブルに無理矢理UPSERTする

背景

PostgreSQLには外部のPostgreSQLサーバに格納されたデータをアクセスするためのpostgres_fdwモジュールがある. これにより,外部のテーブルを自身のサーバにあるかのように扱うことが可能となるが様々な落とし穴がある.

そのひとつは,UPSERT(INSERTON CONFLICT DO UPDATE)が使用できないことである. ドキュメントにも下記の記載がある.

Note that postgres_fdw currently lacks support for INSERT statements with an ON CONFLICT DO UPDATE clause. However, the ON CONFLICT DO NOTHING clause is supported, provided a unique index inference specification is omitted.

しかし,何とかして外部テーブルにUPSERTをしなければならない場面があったので,この記事では代替手法について述べる.

準備

本記事では,下記のテーブルを例にとりUPSERT対象として扱う.

CREATE TABLE t_upsert_test (
  pkey1 VARCHAR(32) NOT NULL
  , pkey2 INTEGER NOT NULL
  , val1 VARCHAR(32) NOT NULL
  , val2 VARCHAR(32) NOT NULL
  , val3 INTEGER
  , val4 INTEGER
  , PRIMARY KEY (pkey1, pkey2)
);

他テーブルからのSELECT結果をUPSERT

通常のUPSERT

PostgreSQLの通常のUPSERT(ON CONFLICT DO NOTHING)は下記の形となる. t_upsert_test_tmp から t_upsert_test_tmp にUPSERTを行う例を示している.

INSERT INTO t_upsert_test (
  pkey1
  , pkey2
  , val1
  , val2
  , val3
  , val4
)
SELECT
  pkey1
  , pkey2
  , val1
  , val2
  , val3
  , val4
FROM
  t_upsert_test_tmp
WHERE
  val3 = 0
  AND val4 = 1
ON CONFLICT (pkey1, pkey2)
DO UPDATE
SET
  val1 = EXCLUDED.val1
  , val2 = EXCLUDED.val2
  , val3 = EXCLUDED.val3
  , val4 = EXCLUDED.val4
;

代替UPSERT

ドキュメントに記載の通り,ON CONFLICT DO NOTHING 句(INSERTできなかったレコードは無視)は使用可能である. PostgreSQLではRETURNING句を用いることで,INSERTしたレコードを取得することができる. これらとWITHと組み合わせることで,何とかUPSERTの代替を実現することができると考えた.

WITH ct1 AS (
  SELECT
    pkey1
    , pkey2
    , val1
    , val2
    , val3
    , val4
  FROM
    t_upsert_test_tmp
  WHERE
    val3 = 0
    AND val4 = 1
)
, ct2 AS (
  INSERT INTO t_upsert_test (
    pkey1
    , pkey2
    , val1
    , val2
    , val3
    , val4
  )
  SELECT
    pkey1
    , pkey2
    , val1
    , val2
    , val3
    , val4
  FROM
    ct1
  ON CONFLICT DO NOTHING
  RETURNING
    *
)
UPDATE t_upsert_test
SET
  val1 = t1.val1
  , val2 = t1.val2
  , val3 = t1.val3
  , val4 = t1.val4
FROM (
  SELECT
    pkey1
    , pkey2
    , val1
    , val2
    , val3
    , val4
  FROM
    ct1
  EXCEPT
  SELECT
    pkey1
    , pkey2
    , val1
    , val2
    , val3
    , val4
  FROM
    ct2
) t1
WHERE
  -- ※1
  t_upsert_test.pkey1 = t1.pkey1
  AND t_upsert_test.pkey2 = t1.pkey2
;

※1は主キーの一致条件を並べるだけのWHERE句である,

やっていることとしては下記の通り.

  1. t_upsert_test_tmp から t_upsert_test にUPSERTしたいレコードを抽出
  2. 抽出したレコードを t_upsert_testINSERTし,主キー重複レコードがあった場合は無視する(ON CONFLICT DO NOTHING
  3. 抽出したレコードとINSERTに成功したレコードの差集合を求め(これがUPDATEすべきレコード),UPDATEを実行する.

CTE(Common Table Expression)はコストが高くなりがちであるが,対象テーブルが外部テーブルであるならばCTEで一度共通化しておいた方がコストは低くなるようだ(上記の例では ct1).

単一のUPSERT

通常のUPSERT

INSERT INTO t_upsert_test (
  pkey1
  , pkey2
  , val1
  , val2
  , val3
  , val4
)
VALUES (
  'ABCD'
  , 42
  , 'aa'
  , 'bb'
  , 1
  , 2
)
ON CONFLICT (pkey1, pkey2)
DO UPDATE
SET
  val1 = EXCLUDED.val1
  , val2 = EXCLUDED.val2
  , val3 = EXCLUDED.val3
  , val4 = EXCLUDED.val4
;

代替UPSERT

冗長な書き方となるが,下記のように代替UPSERTの形にすることができる.

WITH ct AS (
  INSERT INTO t_upsert_test (
    pkey1
    , pkey2
    , val1
    , val2
    , val3
    , val4
  )
  VALUES (
    'ABCD'
    , 42
    , 'aa'
    , 'bb'
    , 1
    , 2
  )
  ON CONFLICT DO NOTHING
  RETURNING
    *
)
UPDATE t_upsert_test
SET
  -- ※1
  val1 = 'aa'
  , val2 = 'bb'
  , val3 = 1
  , val4 = 2
WHERE
  -- ※2
  pkey1 = 'ABCD'
  AND pkey2 = 42
  -- ※3
  AND EXISTS(SELECT * FROM ct)
;

INSERTで指定した値を再度UPDATE部で指定し直さなければならないのが何とも冗長である.

※1はINSERTで指定した値のうち,主キーを除くものを指定している. ※2はINSERTで指定した主キーに対する値を条件としている. ※3はINSERTできたレコードが無かった場合,という条件指定である.

無理矢理1つのクエリにまとめる形としたが,多くの場合SQLはプログラムから発行すると思うので,

  1. INSERTON CONFLICT DO NOTHING)を行う
  2. INSERT件数が0件ならUPDATEを行う

といった形で,INSERTUPDATEを別々に実行した方がよいと思われる.

まとめ

  • PostgreSQLでは外部テーブルに対し,通常のUPSERT(ON CONFLICT DO UPDATE)はできない
  • RETURNINGEXCEPTを組み合わせることで,高コストではあるが外部テーブルへの代替UPSERTを実現することができる

代替UPSERTは通常のUPSERTに比べ,はるかに高コストであるため,件数が少ないテーブルに対してのみ用いるべきである. そもそも,外部テーブルへのUPSERTを行う必要がないように,設計を見直すべきであると思う.

参考

PNGファイル再圧縮ツールを改良した

TL;DR

2020-12-07にも記事を書いたツールだが,当時より機能追加を行ったため,再度記事を書くことにした. 大きくは下記4点が追加となった.

  1. VRMファイルを処理可能になった
  2. 元のPNGファイルの画像フォーマットをそのままにしておくオプションの追加
  3. IDATのデータ部のサイズを指定可能にした
  4. 全チャンクの保存オプションの追加

リポジトリと実行バイナリ配布場所は下記の通り.

f:id:koturn:20210707033537p:plain
再圧縮前後の様子

1. VRMファイルを処理可能になった

VRMファイルをパースし,内部にPNGファイルデータが含まれる場合,再圧縮を行えるようにした. VRMファイル自体は先頭にjsonがあり,その後にバイナリデータが続き,json内のオフセットと長さ情報からデータを切り出すことが出来る形式のファイルとなっている.

なので,

  1. ファイル先頭のjsonをパース
  2. PNGデータを取り出す
  3. PNGデータを再圧縮
  4. 再圧縮したPNGデータをバイナリデータ部に書き戻す
  5. json全体のオフセットと長さ情報をすべて更新
  6. ファイルとして書き出す

という処理でVRMファイル内のPNGファイルデータの再圧縮を実現している.

2. 元のPNGファイルのピクセルデータフォーマットをそのままにしておくオプションの追加

前の記事にも書いていることであるが,再度書く.

これはzopflipng コマンド では存在している --keep-color-type オプションを使用できるようにしたものである. Google公式のリポジトリのものでは,zopflipng.dllからは --keep-color-type に相当する指定を行うことが出来ないが,クローンしたリポジトリではこの指定が可能になるような修正を行っている.

zopflipngは画像データの展開結果が変化しない(ビットマップデータへ展開した結果,元のものと全画素データを比較しても差がない)ならば,ピクセルデータフォーマットを変更することがある. 例えば,

  1. アルファ値が全て255であるARGB32bit形式をRGB24bit形式に変換
  2. 全体として256色以内の色数しか使われていない画像を8bitインデックス画像に変換
  3. 2とは逆に1bpp(2色のインデックス画像,1byteあたり8ピクセル)をRGB24bit画像変換
    • PLTE チャンクの削減によりデータサイズが小さくなることがある

といった変換を行うことがあり得る. 基本的に画像データの見た目に差が生じるわけではないので,問題が生じることはないが,3. の変換が生じた場合,インデックス画像しか取り扱えないアプリケーションで画像データを開けなくなる(特にEdge等のドット絵ツール). そのため,画像データフォーマットを保持する機能は欠かせない.

3. IDATのデータ部のサイズを指定可能にした

zopflipngは少しでもファイルサイズを削減するために,PNGファイルのIDATチャンクを1つにまとめる(IDATチャンクのデータ部ノサイズは自由であるため,1まとめにすることも分割することも可能.ただし,IDATチャンク1つにつき,データ長,チャンク種別,CRC-32の分のサイズオーバーヘッドが12 Bytesある). しかし,人によってはIDATチャンクをデータ部のサイズが最大8192 Bytesになるように分割したい,あるいはWindowsのpaint.exeのように65535 Bytes近くになるように分割したいという人もいるかもしれない.

そのため,PNGデータの再圧縮完了後にIDATチャンクを分割する処理を行うことができるように,オプション --idat-size を追加した.

このオプションを指定しなかったり,0やマイナスの値を指定すると,今まで通りIDATの分割処理を行わないが,例えば --idat-size=8192 と指定するとIDATチャンクのデータ部がそれぞれ最大8192 BytesとなるようにIDATチャンクを分割する.

IDATチャンクを小さくすることで,PNGデコーダの読み取り時の一時バッファのサイズを小さく保つことが出来るようになるが,これが速度面に寄与するかどうかは不明である. PNGの仕様としては下記の記述があるだけで,C言語的な時代を感じる.

Multiple IDAT chunks are allowed so that encoders can work in a fixed amount of memory; typically the chunk size will correspond to the encoder's buffer size.

VRChatやclusterの画像データ,またはGIMPといったアプリケーションが生成するPNGファイルのIDATチャンクのデータ部のサイズとしては 8192 Bytes が採用されている. これはおそらくlibpngのデフォルトのIDATのサイズなのではないかと思う.

4. 全チャンクの保存オプションの追加

--keep-all-chunks というオプションを指定すると,すべてのチャンクを保持したままにするようにした. このオプションは --keep-chunks=acTL,bKGD,cHRM,eXIf,fcTL,fdAT,gAMA,hIST,iCCP,iTXt,pHYs,sBIT,sPLT,sRGB,tEXt,tIME,zTXt を指定したのと同じ効果である.

zopflipngはファイルサイズを小さくすることを最優先にしているので,必須ではないチャンクは基本的に除去する. 以前のものでも保持したいチャンクは --keep-chunks= で指定することで残すことが出来たが,雑にメタデータチャンクを残したい場合にでも1つ1つチャンク名を指定しなければならないのは面倒であったため,このオプションを追加した.

裏で改善したところ

以下の項目は特に機能追加ではなく,ほとんどが自己満足的な改良である.

.NET 5への移行

最初は.NET Framework 4.8で作成していたが,最新のC#と標準ライブラリ(SpanSIMDAPI)を利用したかったので,.NET 5に移行した.

.NET 5では,.NET 5依存のバイナリをそのままリリースすることもできるし,Self-Containedという実行環境にて.NET 5が不要になるバイナリの作成を行うこともできる(その分ファイルサイズがかなり大きくなるが).

再圧縮結果のアンマネージドメモリをそのまま使うようにした

以前は koturn/ZopfliSharp にて,zopflipng.dll の関数呼び出しより得たアンマネージドメモリはすぐさま同サイズのマネージドメモリを確保し,そこにコピーし,そのマネージドメモリを返却するようにしていた

よく知られているように,.NETでは85KB以上のメモリ確保を行うと,LOHに確保されてしまう. 僕がよく再圧縮処理対象にするVRChatやclusterのPNGファイルはサイズが大きいため,再圧縮結果が85KB以内に収まるということはあまりない(VRChatの画像の多くは1.5MB,clusterの画像は1MB弱程度になる). そのため,このメモリアロケーションが気になっていた.

そこで,アンマネージドメモリを SafeBuffer として返却するメソッドをkoturn/ZopfliSharp追加し,それを利用するようにした. この変更により,処理時間が劇的に短くなるわけではないが,P/Invoke用のライブラリを作った以上はこのメソッドは欲しかったのと,趣味プロダクトであるので,微々たる処理の最適化を行いたかった.

再圧縮結果がオリジナルのものより悪ければ,オリジナルの方をそのまま使うという処理のために byte 配列と SafeBuffer を混在して扱う必要が生じたが,前述の通り,.NET 5へ移行したので,Span を利用し,統一的に扱うことによってコードをシンプルに保った. SafeBuffer から Span を生成するのにあたって下記のようにした.

private static unsafe Span<byte> CreateSpan(SafeBuffer sb)
{
    return new Span(sb.DangerousGetHandle(), (int)sb.ByteLength);
}

また,再圧縮のVerificationに必要な Bitmap インスタンスの生成にPNG画像データの Stream が必要になるが,UnmanagedMemoryStream を利用すれば,byte 配列に対する MemoryStream と同じノリで扱うことができる.

private static Bitmap CreateBitmap(SafeBuffer sb)
{
    using ums = new UnmanagedMemoryStream(sb, 0, (long)sb.ByteLength);
    return (Bitmap)Image.FromStream(ums);
}

SafeBufferDispose() メソッドの呼び出しで,中身のアンマネージドメモリを解放することができるので,using句を用いればメモリ解放漏れは無くなる.

using (var sb = Zopfli.OptimizePngUnmanaged(data, 0, data.Length))
{
    // 処理
}

画像比較ログの改善

zopflipngによる再圧縮のVerificationとして,元の画像と比較する機能がある. 以前は画像フォーマットが異なる場合,画像に差があることしかわからないログであったが,どのように画像フォーマットが変化したかわかるようにログ出力するようにした.

参考

clusterの写真リネームツールを作った

TL;DR

clusterで撮影した画像ファイルの一括ダウンロードで得られるzipファイル内のPNGファイル名を cluster_yyyy-MM-dd_HH-mm-ss_XXX.png という形式にリネームするツールを作った(yyyyMMddHHmmss は撮影日時の年月日時分秒,撮影日時=ファイル作成日時=ファイル最終更新時刻). ついでに,PNGファイルにメタ情報として,ファイル撮影日時を埋め込むようにした.

背景

clusterのPNG画像はGUID形式のもので整理するのに不便という話がいつもたむろしているDiscordのチャットにあった. 複数選択して一括ダウンロードしたPNG画像はタイムスタンプが保持されていることは知っていたので,zipファイルを突っ込んだら中のファイルをリネームして,新しいzipファイルを作るCLIツールを作成した. また,画像のメタデータとして日付とかを持てるといいかもという話もあったので,ついでにそれも追加するようにした.

(ファイル名がGUIDであるのは,DBに突っ込む上での都合なのかどうかという想像が出来て楽しいところ)

PNGファイルの構造

PNGファイルの構造は非常に単純であり,先頭8バイトにシグネチャがあり,そこから下記の形式のチャンクが続くだけである. ただし,データ長やCRC-32といった数値はビッグエンディアンであることに注意しなければならない.

サイズ 意味
4 Bytes チャンクデータ長(N)
4 Bytes チャンク種別を示すASCIIテキスト4文字(IHDR, IDAT, tEXt など)
N Bytes チャンクに応じたデータ部
4 Bytes チャンク種別とデータ部のCRC-32

最初にIHDR,最後にIEND,IDATチャンクが複数ある場合はIDATチャンクの間に他のチャンクが存在してはならず,データ順になるように連続していなければならない等の制約があるが,全体的な構造を示す部分はなくただチャンクが並んでいるだけなので,チャンクの追加は容易である.

チャンク種別のASCII文字の何文字目が大文字か小文字かによって,チャンクの特性を表現している.

文字 意味
1文字目 大文字であれば必須チャンク
2文字目 大文字であれば仕様が公開・定義されているもの
3文字目 将来のために予約されているが,現在は常に大文字
4文字目 大文字の場合,他の必須チャンクの影響を受けるので,そのままコピーはできない

tEXtチャンク

tEXtチャンクの構造は下記の通り. データ部がASCII文字でキーと値がNULL文字で区切られている構造となっている. 仕様としては,キー部は80文字以内という但し書きがある.

サイズ 意味
4 Bytes データ長(M + 1 + N)
4 Bytes tEXt
M Bytes キー文字列
1 Byte \0 (NULL文字)
N Bytes 値となる文字列
4 Bytes CRC-32

Creation Time

WindowsエクスプローラではPNGファイルにキー:Creation Time,値:yyyy:MM:dd HH:mm 形式の時刻文字列のtEXtチャンクが存在する場合,それを撮影日時として表示する仕様となっているようだ. そのため,PNGファイルにキー:Creation Time,値:ファイル最終更新時刻をyyyy:MM:dd HH:mm:ss形式にした文字列 を埋め込む機能を入れた. これにより,ファイル名を手動で再度リネームしても更新時刻は失われなくなる. なお,ファイル1つにつき,31 Bytesサイズが増加するが微々たるものであるため,良しとした.

PNGの仕様としては時刻文字列としては RFC 1123 が推奨されるようだが,利便性を考え,Windowsエクスプローラ形式を採用した.

Title

ファイルをリネームするにあたって,もともとのファイル名もメタデータとして残すことにした. というのも,clusterのPNGファイル名はGUID形式のものであり,重複は基本的にない一意のものであると見做せるためだ. だからといって何かになるわけではないが,今後何かあったときに役立てることが出来る可能性がある.

tIMEチャンク

PNGファイルの仕様として,最終更新時刻を保持するためのチャンクも用意されている. これを設定したからといって,エクスプローラの表示に影響を与えるわけではないが,ついでなので追加することにした. UTC(あるいはGMT)の時刻が推奨されるため,UTCに変更して保存するようにした. そのため,値だけ見ればCreation Timeに保存した時刻と9時間のズレがある.

サイズ 意味
4 Bytes データ長(7)
4 Bytes tIME
2 Bytes
1 Byte
1 Byte
1 Byte
1 Byte
1 Byte
4 Bytes CRC-32

zipファイルの時刻精度について

zipファイルの仕様上,zipファイル内のファイルの時刻精度は2秒刻みである. そのため,時刻を含むだけの形式では,ファイル名重複が容易に起こりうるため,連番部分を追加している. 敢えてミリ秒っぽく3桁分を用意しているが,clusterではどう努力しても2秒以内に100枚以上のファイルどころか10枚以上のファイルの保存はできないと思われるので,1桁でもよいのではないかと思ってはいる.

あえてやらなかったこと

clusterの写真ファイルは保存速度のため,圧縮率レベルは1にして保存しているらしい(これはVRChatの写真も同様). (PNGの画像データはzlib形式のDeflate圧縮を採用しているので,実際に伸長して再圧縮して確かめてみた感じだと,zlibの最低の圧縮レベル圧縮率を指定したときと同一の結果となった) なので,ある程度の圧縮レベルにして,再圧縮するようにすれば,追加したチャンク分以上にデータサイズを削ることが出来ると考えた. だが,あくまでリネームツールのため,再圧縮は範疇外として採用しなかった. (再圧縮の処理時間が1秒だとしても対象ファイルが600枚あれば10分はかかってしまう)

同様の理由でIDATチャンクの結合も行わないことにしている(clusterのPNG画像はIDATのデータ部を 8192 Bytes 毎になるように分割している.IDATを結合することで,IDAT1つあたり 12 Bytes(データ長 + チャンク種別 + CRC-32) のファイルサイズを削減することが出来る).

再圧縮については,別で作成したPNGファイルの超圧縮ツールの役割であると思う.

参考

zipファイルの再圧縮ツールを作った

TL;DR

zopfliを用いて,zipファイルを再圧縮し,よりよい圧縮結果を得るCLIツールを作成した.

元ネタはkomiya-atsushi/zipzopであり,これをC# で再実装したものとなるが,

  1. ファイル単位での並列処理が可能
  2. zopfli.dll の関数に渡せるオプションを全て指定可能

という点が異なる.

背景

zopflipngを用いたPNG再圧縮ツールを作成した際に,「多くのzipファイルにもDeflate圧縮が用いられているのだから,zopfliでzipファイルを再圧縮できるはずだ」と考えていた.

探してみると,komiya-atsushi/zipzopというリポジトリが見つかった. しかし,zipファイル内のPNGファイルを並列で再圧縮するツールを作っていた身としては,並列処理できない点を不満に思った.

zopfliのP/Invoke用のライブラリは作成していたので,とりあえずC# で上記のツールを実装することにした.

使用方法

下記の通り. ディレクトリ指定時はそのディレクトリをzipファイルに圧縮した後,再圧縮を行う.

> RecompressZip.exe 【オプション】... 【zipファイル or ディレクトリ】...

オプション

オプション 機能
-b 【数値】, --block-split-max=【数値】 最大のブロック分割数を指定する.0は無制限を意味する.デフォルトは15.
-d, --dry-run 圧縮処理は行うが,ファイル内容の置き換えは行わない.ベンチマーク用のオプション.
-h, --help 使い方を表示し,プログラムを終了する.
-i 【数値】, --num-iteration=【数値】 最大の繰り返し回数を指定する.デフォルトは15.
-r, --replace-force 再圧縮の結果がオリジナルよりも悪かったとしても置き換えを行うようにする.
-v, --verbose zopfli.dllからの標準エラー出力へのデバッグ出力を有効にする.
-V, --verbose-more zopfli.dllからの標準エラー出力へのより詳細なデバッグ出力を有効にする.
--no-block-split ブロック分割を行わないようにする.
--no-overwrite 元のzipファイルの置き換えを行わないようにする.

使用した所感

7zipで作成した最高レベルの圧縮率のzipファイルに対して使用してみたところ,おおよそ1つのファイル(zipファイルエントリ)につき1%弱程度より圧縮することができた. しかし,ファイルによっては7zipの方が結果が良いこともあり,zopfliが万能ということでもなさそうだ.

普段はzipファイルを作成するときには,7zipで作っておいて,数バイトでも小さくしたいならば,本ツールにて再圧縮をかけるとよいと思う(-r を指定しない限り,元のものより悪い結果は採用しない).

または,Excelファイル(.xlsx)のようなアプリケーション自体がさほど圧縮率を気にせずに作成したzipファイルに対して,このツールを用いるとよいと思われる.

制限事項

元ネタをそのまま実装した形なので,下記の通り. これらについては特に対応することを考えていない.

  • 暗号化zip非対応
  • Deflate圧縮以外はそのまま格納(Deflate64も)

参考